ceph/src/rgw/rgw_rest_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_rest_log.h"
  20 #include "rgw_client_io.h"
  21 #include "rgw_sync.h"
  22 #include "rgw_data_sync.h"
  23 #include "rgw_common.h"
  24 #include "common/errno.h"
  25 #include "include/assert.h"
  26
  27 #define dout_context g_ceph_context
  28 #define LOG_CLASS_LIST_MAX_ENTRIES (1000)
  29 #define dout_subsys ceph_subsys_rgw
  30
  31 static int parse_date_str(string& in, real_time& out) {
  32   uint64_t epoch = 0;
  33   uint64_t nsec = 0;
  34
  35   if (!in.empty()) {
  36     if (utime_t::parse_date(in, &epoch, &nsec) < 0) {
  37       dout(5) << "Error parsing date " << in << dendl;
  38       return -EINVAL;
  39     }
  40   }
  41   out = utime_t(epoch, nsec).to_real_time();
  42   return 0;
  43 }
  44
  45 void RGWOp_MDLog_List::execute() {
  46   string   period = s->info.args.get("period");
  47   string   shard = s->info.args.get("id");
  48   string   max_entries_str = s->info.args.get("max-entries");
  49   string   st = s->info.args.get("start-time"),
  50            et = s->info.args.get("end-time"),
  51            marker = s->info.args.get("marker"),
  52            err;
  53   real_time  ut_st,
  54              ut_et;
  55   void    *handle;
  56   unsigned shard_id, max_entries = LOG_CLASS_LIST_MAX_ENTRIES;
  57
  58   shard_id = (unsigned)strict_strtol(shard.c_str(), 10, &err);
  59   if (!err.empty()) {
  60     dout(5) << "Error parsing shard_id " << shard << dendl;
  61     http_ret = -EINVAL;
  62     return;
  63   }
  64
  65   if (parse_date_str(st, ut_st) < 0) {
  66     http_ret = -EINVAL;
  67     return;
  68   }
  69
  70   if (parse_date_str(et, ut_et) < 0) {
  71     http_ret = -EINVAL;
  72     return;
  73   }
  74
  75   if (!max_entries_str.empty()) {
  76     max_entries = (unsigned)strict_strtol(max_entries_str.c_str(), 10, &err);
  77     if (!err.empty()) {
  78       dout(5) << "Error parsing max-entries " << max_entries_str << dendl;
  79       http_ret = -EINVAL;
  80       return;
  81     }
  82     if (max_entries > LOG_CLASS_LIST_MAX_ENTRIES) {
  83       max_entries = LOG_CLASS_LIST_MAX_ENTRIES;
  84     }
  85   }
  86
  87   if (period.empty()) {
  88     ldout(s->cct, 5) << "Missing period id trying to use current" << dendl;
  89     period = store->get_current_period_id();
  90     if (period.empty()) {
  91       ldout(s->cct, 5) << "Missing period id" << dendl;
  92       http_ret = -EINVAL;
  93       return;
  94     }
  95   }
  96
  97   RGWMetadataLog meta_log{s->cct, store, period};
  98
  99   meta_log.init_list_entries(shard_id, ut_st, ut_et, marker, &handle);
 100
 101   http_ret = meta_log.list_entries(handle, max_entries, entries,
 102                                    &last_marker, &truncated);
 103
 104   meta_log.complete_list_entries(handle);
 105 }
 106
 107 void RGWOp_MDLog_List::send_response() {
 108   set_req_state_err(s, http_ret);
 109   dump_errno(s);
 110   end_header(s);
 111
 112   if (http_ret < 0)
 113     return;
 114
 115   s->formatter->open_object_section("log_entries");
 116   s->formatter->dump_string("marker", last_marker);
 117   s->formatter->dump_bool("truncated", truncated);
 118   {
 119     s->formatter->open_array_section("entries");
 120     for (list<cls_log_entry>::iterator iter = entries.begin();
 121          iter != entries.end(); ++iter) {
 122       cls_log_entry& entry = *iter;
 123       store->meta_mgr->dump_log_entry(entry, s->formatter);
 124       flusher.flush();
 125     }
 126     s->formatter->close_section();
 127   }
 128   s->formatter->close_section();
 129   flusher.flush();
 130 }
 131
 132 void RGWOp_MDLog_Info::execute() {
 133   num_objects = s->cct->_conf->rgw_md_log_max_shards;
 134   period = store->meta_mgr->read_oldest_log_period();
 135   http_ret = period.get_error();
 136 }
 137
 138 void RGWOp_MDLog_Info::send_response() {
 139   set_req_state_err(s, http_ret);
 140   dump_errno(s);
 141   end_header(s);
 142
 143   s->formatter->open_object_section("mdlog");
 144   s->formatter->dump_unsigned("num_objects", num_objects);
 145   if (period) {
 146     s->formatter->dump_string("period", period.get_period().get_id());
 147     s->formatter->dump_unsigned("realm_epoch", period.get_epoch());
 148   }
 149   s->formatter->close_section();
 150   flusher.flush();
 151 }
 152
 153 void RGWOp_MDLog_ShardInfo::execute() {
 154   string period = s->info.args.get("period");
 155   string shard = s->info.args.get("id");
 156   string err;
 157
 158   unsigned shard_id = (unsigned)strict_strtol(shard.c_str(), 10, &err);
 159   if (!err.empty()) {
 160     dout(5) << "Error parsing shard_id " << shard << dendl;
 161     http_ret = -EINVAL;
 162     return;
 163   }
 164
 165   if (period.empty()) {
 166     ldout(s->cct, 5) << "Missing period id trying to use current" << dendl;
 167     period = store->get_current_period_id();
 168
 169     if (period.empty()) {
 170       ldout(s->cct, 5) << "Missing period id" << dendl;
 171       http_ret = -EINVAL;
 172       return;
 173     }
 174   }
 175   RGWMetadataLog meta_log{s->cct, store, period};
 176
 177   http_ret = meta_log.get_info(shard_id, &info);
 178 }
 179
 180 void RGWOp_MDLog_ShardInfo::send_response() {
 181   set_req_state_err(s, http_ret);
 182   dump_errno(s);
 183   end_header(s);
 184
 185   encode_json("info", info, s->formatter);
 186   flusher.flush();
 187 }
 188
 189 void RGWOp_MDLog_Delete::execute() {
 190   string   st = s->info.args.get("start-time"),
 191            et = s->info.args.get("end-time"),
 192            start_marker = s->info.args.get("start-marker"),
 193            end_marker = s->info.args.get("end-marker"),
 194            period = s->info.args.get("period"),
 195            shard = s->info.args.get("id"),
 196            err;
 197   real_time  ut_st,
 198              ut_et;
 199   unsigned shard_id;
 200
 201   http_ret = 0;
 202
 203   shard_id = (unsigned)strict_strtol(shard.c_str(), 10, &err);
 204   if (!err.empty()) {
 205     dout(5) << "Error parsing shard_id " << shard << dendl;
 206     http_ret = -EINVAL;
 207     return;
 208   }
 209   if (et.empty() && end_marker.empty()) { /* bounding end */
 210     http_ret = -EINVAL;
 211     return;
 212   }
 213
 214   if (parse_date_str(st, ut_st) < 0) {
 215     http_ret = -EINVAL;
 216     return;
 217   }
 218
 219   if (parse_date_str(et, ut_et) < 0) {
 220     http_ret = -EINVAL;
 221     return;
 222   }
 223
 224   if (period.empty()) {
 225     ldout(s->cct, 5) << "Missing period id trying to use current" << dendl;
 226     period = store->get_current_period_id();
 227
 228     if (period.empty()) {
 229       ldout(s->cct, 5) << "Missing period id" << dendl;
 230       http_ret = -EINVAL;
 231       return;
 232     }
 233   }
 234   RGWMetadataLog meta_log{s->cct, store, period};
 235
 236   http_ret = meta_log.trim(shard_id, ut_st, ut_et, start_marker, end_marker);
 237 }
 238
 239 void RGWOp_MDLog_Lock::execute() {
 240   string period, shard_id_str, duration_str, locker_id, zone_id;
 241   unsigned shard_id;
 242
 243   http_ret = 0;
 244
 245   period       = s->info.args.get("period");
 246   shard_id_str = s->info.args.get("id");
 247   duration_str = s->info.args.get("length");
 248   locker_id    = s->info.args.get("locker-id");
 249   zone_id      = s->info.args.get("zone-id");
 250
 251   if (period.empty()) {
 252     ldout(s->cct, 5) << "Missing period id trying to use current" << dendl;
 253     period = store->get_current_period_id();
 254   }
 255
 256   if (period.empty() ||
 257       shard_id_str.empty() ||
 258       (duration_str.empty()) ||
 259       locker_id.empty() ||
 260       zone_id.empty()) {
 261     dout(5) << "Error invalid parameter list" << dendl;
 262     http_ret = -EINVAL;
 263     return;
 264   }
 265
 266   string err;
 267   shard_id = (unsigned)strict_strtol(shard_id_str.c_str(), 10, &err);
 268   if (!err.empty()) {
 269     dout(5) << "Error parsing shard_id param " << shard_id_str << dendl;
 270     http_ret = -EINVAL;
 271     return;
 272   }
 273
 274   RGWMetadataLog meta_log{s->cct, store, period};
 275   unsigned dur;
 276   dur = (unsigned)strict_strtol(duration_str.c_str(), 10, &err);
 277   if (!err.empty() || dur <= 0) {
 278     dout(5) << "invalid length param " << duration_str << dendl;
 279     http_ret = -EINVAL;
 280     return;
 281   }
 282   http_ret = meta_log.lock_exclusive(shard_id, make_timespan(dur), zone_id,
 283                                      locker_id);
 284   if (http_ret == -EBUSY)
 285     http_ret = -ERR_LOCKED;
 286 }
 287
 288 void RGWOp_MDLog_Unlock::execute() {
 289   string period, shard_id_str, locker_id, zone_id;
 290   unsigned shard_id;
 291
 292   http_ret = 0;
 293
 294   period       = s->info.args.get("period");
 295   shard_id_str = s->info.args.get("id");
 296   locker_id    = s->info.args.get("locker-id");
 297   zone_id      = s->info.args.get("zone-id");
 298
 299   if (period.empty()) {
 300     ldout(s->cct, 5) << "Missing period id trying to use current" << dendl;
 301     period = store->get_current_period_id();
 302   }
 303
 304   if (period.empty() ||
 305       shard_id_str.empty() ||
 306       locker_id.empty() ||
 307       zone_id.empty()) {
 308     dout(5) << "Error invalid parameter list" << dendl;
 309     http_ret = -EINVAL;
 310     return;
 311   }
 312
 313   string err;
 314   shard_id = (unsigned)strict_strtol(shard_id_str.c_str(), 10, &err);
 315   if (!err.empty()) {
 316     dout(5) << "Error parsing shard_id param " << shard_id_str << dendl;
 317     http_ret = -EINVAL;
 318     return;
 319   }
 320
 321   RGWMetadataLog meta_log{s->cct, store, period};
 322   http_ret = meta_log.unlock(shard_id, zone_id, locker_id);
 323 }
 324
 325 void RGWOp_MDLog_Notify::execute() {
 326   char *data;
 327   int len = 0;
 328 #define LARGE_ENOUGH_BUF (128 * 1024)
 329   int r = rgw_rest_read_all_input(s, &data, &len, LARGE_ENOUGH_BUF);
 330   if (r < 0) {
 331     http_ret = r;
 332     return;
 333   }
 334
 335   ldout(s->cct, 20) << __func__ << "(): read data: " << string(data, len) << dendl;
 336
 337   JSONParser p;
 338   r = p.parse(data, len);
 339   free(data);
 340   if (r < 0) {
 341     ldout(s->cct, 0) << "ERROR: failed to parse JSON" << dendl;
 342     http_ret = r;
 343     return;
 344   }
 345
 346   set<int> updated_shards;
 347   try {
 348     decode_json_obj(updated_shards, &p);
 349   } catch (JSONDecoder::err& err) {
 350     ldout(s->cct, 0) << "ERROR: failed to decode JSON" << dendl;
 351     http_ret = -EINVAL;
 352     return;
 353   }
 354
 355   if (store->ctx()->_conf->subsys.should_gather(ceph_subsys_rgw, 20)) {
 356     for (set<int>::iterator iter = updated_shards.begin(); iter != updated_shards.end(); ++iter) {
 357       ldout(s->cct, 20) << __func__ << "(): updated shard=" << *iter << dendl;
 358     }
 359   }
 360
 361   store->wakeup_meta_sync_shards(updated_shards);
 362
 363   http_ret = 0;
 364 }
 365
 366 void RGWOp_BILog_List::execute() {
 367   string tenant_name = s->info.args.get("tenant"),
 368          bucket_name = s->info.args.get("bucket"),
 369          marker = s->info.args.get("marker"),
 370          max_entries_str = s->info.args.get("max-entries"),
 371          bucket_instance = s->info.args.get("bucket-instance");
 372   RGWBucketInfo bucket_info;
 373   unsigned max_entries;
 374
 375   RGWObjectCtx& obj_ctx = *static_cast<RGWObjectCtx *>(s->obj_ctx);
 376
 377   if (bucket_name.empty() && bucket_instance.empty()) {
 378     dout(5) << "ERROR: neither bucket nor bucket instance specified" << dendl;
 379     http_ret = -EINVAL;
 380     return;
 381   }
 382
 383   int shard_id;
 384   http_ret = rgw_bucket_parse_bucket_instance(bucket_instance, &bucket_instance, &shard_id);
 385   if (http_ret < 0) {
 386     return;
 387   }
 388
 389   if (!bucket_instance.empty()) {
 390     http_ret = store->get_bucket_instance_info(obj_ctx, bucket_instance, bucket_info, NULL, NULL);
 391     if (http_ret < 0) {
 392       dout(5) << "could not get bucket instance info for bucket instance id=" << bucket_instance << dendl;
 393       return;
 394     }
 395   } else { /* !bucket_name.empty() */
 396     http_ret = store->get_bucket_info(obj_ctx, tenant_name, bucket_name, bucket_info, NULL, NULL);
 397     if (http_ret < 0) {
 398       dout(5) << "could not get bucket info for bucket=" << bucket_name << dendl;
 399       return;
 400     }
 401   }
 402
 403   bool truncated;
 404   unsigned count = 0;
 405   string err;
 406
 407   max_entries = (unsigned)strict_strtol(max_entries_str.c_str(), 10, &err);
 408   if (!err.empty())
 409     max_entries = LOG_CLASS_LIST_MAX_ENTRIES;
 410
 411   send_response();
 412   do {
 413     list<rgw_bi_log_entry> entries;
 414     int ret = store->list_bi_log_entries(bucket_info, shard_id,
 415                                           marker, max_entries - count,
 416                                           entries, &truncated);
 417     if (ret < 0) {
 418       dout(5) << "ERROR: list_bi_log_entries()" << dendl;
 419       return;
 420     }
 421
 422     count += entries.size();
 423
 424     send_response(entries, marker);
 425   } while (truncated && count < max_entries);
 426
 427   send_response_end();
 428 }
 429
 430 void RGWOp_BILog_List::send_response() {
 431   if (sent_header)
 432     return;
 433
 434   set_req_state_err(s, http_ret);
 435   dump_errno(s);
 436   end_header(s);
 437
 438   sent_header = true;
 439
 440   if (http_ret < 0)
 441     return;
 442
 443   s->formatter->open_array_section("entries");
 444 }
 445
 446 void RGWOp_BILog_List::send_response(list<rgw_bi_log_entry>& entries, string& marker)
 447 {
 448   for (list<rgw_bi_log_entry>::iterator iter = entries.begin(); iter != entries.end(); ++iter) {
 449     rgw_bi_log_entry& entry = *iter;
 450     encode_json("entry", entry, s->formatter);
 451
 452     marker = entry.id;
 453     flusher.flush();
 454   }
 455 }
 456
 457 void RGWOp_BILog_List::send_response_end() {
 458   s->formatter->close_section();
 459   flusher.flush();
 460 }
 461
 462 void RGWOp_BILog_Info::execute() {
 463   string tenant_name = s->info.args.get("tenant"),
 464          bucket_name = s->info.args.get("bucket"),
 465          bucket_instance = s->info.args.get("bucket-instance");
 466   RGWBucketInfo bucket_info;
 467
 468   RGWObjectCtx& obj_ctx = *static_cast<RGWObjectCtx *>(s->obj_ctx);
 469
 470   if (bucket_name.empty() && bucket_instance.empty()) {
 471     dout(5) << "ERROR: neither bucket nor bucket instance specified" << dendl;
 472     http_ret = -EINVAL;
 473     return;
 474   }
 475
 476   int shard_id;
 477   http_ret = rgw_bucket_parse_bucket_instance(bucket_instance, &bucket_instance, &shard_id);
 478   if (http_ret < 0) {
 479     return;
 480   }
 481
 482   if (!bucket_instance.empty()) {
 483     http_ret = store->get_bucket_instance_info(obj_ctx, bucket_instance, bucket_info, NULL, NULL);
 484     if (http_ret < 0) {
 485       dout(5) << "could not get bucket instance info for bucket instance id=" << bucket_instance << dendl;
 486       return;
 487     }
 488   } else { /* !bucket_name.empty() */
 489     http_ret = store->get_bucket_info(obj_ctx, tenant_name, bucket_name, bucket_info, NULL, NULL);
 490     if (http_ret < 0) {
 491       dout(5) << "could not get bucket info for bucket=" << bucket_name << dendl;
 492       return;
 493     }
 494   }
 495   map<RGWObjCategory, RGWStorageStats> stats;
 496   int ret =  store->get_bucket_stats(bucket_info, shard_id, &bucket_ver, &master_ver, stats, &max_marker);
 497   if (ret < 0 && ret != -ENOENT) {
 498     http_ret = ret;
 499     return;
 500   }
 501 }
 502
 503 void RGWOp_BILog_Info::send_response() {
 504   set_req_state_err(s, http_ret);
 505   dump_errno(s);
 506   end_header(s);
 507
 508   if (http_ret < 0)
 509     return;
 510
 511   s->formatter->open_object_section("info");
 512   encode_json("bucket_ver", bucket_ver, s->formatter);
 513   encode_json("master_ver", master_ver, s->formatter);
 514   encode_json("max_marker", max_marker, s->formatter);
 515   s->formatter->close_section();
 516
 517   flusher.flush();
 518 }
 519
 520 void RGWOp_BILog_Delete::execute() {
 521   string tenant_name = s->info.args.get("tenant"),
 522          bucket_name = s->info.args.get("bucket"),
 523          start_marker = s->info.args.get("start-marker"),
 524          end_marker = s->info.args.get("end-marker"),
 525          bucket_instance = s->info.args.get("bucket-instance");
 526
 527   RGWBucketInfo bucket_info;
 528
 529   RGWObjectCtx& obj_ctx = *static_cast<RGWObjectCtx *>(s->obj_ctx);
 530
 531   http_ret = 0;
 532   if ((bucket_name.empty() && bucket_instance.empty()) ||
 533       end_marker.empty()) {
 534     dout(5) << "ERROR: one of bucket and bucket instance, and also end-marker is mandatory" << dendl;
 535     http_ret = -EINVAL;
 536     return;
 537   }
 538
 539   int shard_id;
 540   http_ret = rgw_bucket_parse_bucket_instance(bucket_instance, &bucket_instance, &shard_id);
 541   if (http_ret < 0) {
 542     return;
 543   }
 544
 545   if (!bucket_instance.empty()) {
 546     http_ret = store->get_bucket_instance_info(obj_ctx, bucket_instance, bucket_info, NULL, NULL);
 547     if (http_ret < 0) {
 548       dout(5) << "could not get bucket instance info for bucket instance id=" << bucket_instance << dendl;
 549       return;
 550     }
 551   } else { /* !bucket_name.empty() */
 552     http_ret = store->get_bucket_info(obj_ctx, tenant_name, bucket_name, bucket_info, NULL, NULL);
 553     if (http_ret < 0) {
 554       dout(5) << "could not get bucket info for bucket=" << bucket_name << dendl;
 555       return;
 556     }
 557   }
 558   http_ret = store->trim_bi_log_entries(bucket_info, shard_id, start_marker, end_marker);
 559   if (http_ret < 0) {
 560     dout(5) << "ERROR: trim_bi_log_entries() " << dendl;
 561   }
 562   return;
 563 }
 564
 565 void RGWOp_DATALog_List::execute() {
 566   string   shard = s->info.args.get("id");
 567
 568   string   st = s->info.args.get("start-time"),
 569            et = s->info.args.get("end-time"),
 570            max_entries_str = s->info.args.get("max-entries"),
 571            marker = s->info.args.get("marker"),
 572            err;
 573   real_time  ut_st,
 574              ut_et;
 575   unsigned shard_id, max_entries = LOG_CLASS_LIST_MAX_ENTRIES;
 576
 577   s->info.args.get_bool("extra-info", &extra_info, false);
 578
 579   shard_id = (unsigned)strict_strtol(shard.c_str(), 10, &err);
 580   if (!err.empty()) {
 581     dout(5) << "Error parsing shard_id " << shard << dendl;
 582     http_ret = -EINVAL;
 583     return;
 584   }
 585
 586   if (parse_date_str(st, ut_st) < 0) {
 587     http_ret = -EINVAL;
 588     return;
 589   }
 590
 591   if (parse_date_str(et, ut_et) < 0) {
 592     http_ret = -EINVAL;
 593     return;
 594   }
 595
 596   if (!max_entries_str.empty()) {
 597     max_entries = (unsigned)strict_strtol(max_entries_str.c_str(), 10, &err);
 598     if (!err.empty()) {
 599       dout(5) << "Error parsing max-entries " << max_entries_str << dendl;
 600       http_ret = -EINVAL;
 601       return;
 602     }
 603     if (max_entries > LOG_CLASS_LIST_MAX_ENTRIES) {
 604       max_entries = LOG_CLASS_LIST_MAX_ENTRIES;
 605     }
 606   }
 607
 608   // Note that last_marker is updated to be the marker of the last
 609   // entry listed
 610   http_ret = store->data_log->list_entries(shard_id, ut_st, ut_et,
 611                                            max_entries, entries, marker,
 612                                            &last_marker, &truncated);
 613 }
 614
 615 void RGWOp_DATALog_List::send_response() {
 616   set_req_state_err(s, http_ret);
 617   dump_errno(s);
 618   end_header(s);
 619
 620   if (http_ret < 0)
 621     return;
 622
 623   s->formatter->open_object_section("log_entries");
 624   s->formatter->dump_string("marker", last_marker);
 625   s->formatter->dump_bool("truncated", truncated);
 626   {
 627     s->formatter->open_array_section("entries");
 628     for (list<rgw_data_change_log_entry>::iterator iter = entries.begin();
 629          iter != entries.end(); ++iter) {
 630       rgw_data_change_log_entry& entry = *iter;
 631       if (!extra_info) {
 632         encode_json("entry", entry.entry, s->formatter);
 633       } else {
 634         encode_json("entry", entry, s->formatter);
 635       }
 636       flusher.flush();
 637     }
 638     s->formatter->close_section();
 639   }
 640   s->formatter->close_section();
 641   flusher.flush();
 642 }
 643
 644
 645 void RGWOp_DATALog_Info::execute() {
 646   num_objects = s->cct->_conf->rgw_data_log_num_shards;
 647   http_ret = 0;
 648 }
 649
 650 void RGWOp_DATALog_Info::send_response() {
 651   set_req_state_err(s, http_ret);
 652   dump_errno(s);
 653   end_header(s);
 654
 655   s->formatter->open_object_section("num_objects");
 656   s->formatter->dump_unsigned("num_objects", num_objects);
 657   s->formatter->close_section();
 658   flusher.flush();
 659 }
 660
 661 void RGWOp_DATALog_ShardInfo::execute() {
 662   string shard = s->info.args.get("id");
 663   string err;
 664
 665   unsigned shard_id = (unsigned)strict_strtol(shard.c_str(), 10, &err);
 666   if (!err.empty()) {
 667     dout(5) << "Error parsing shard_id " << shard << dendl;
 668     http_ret = -EINVAL;
 669     return;
 670   }
 671
 672   http_ret = store->data_log->get_info(shard_id, &info);
 673 }
 674
 675 void RGWOp_DATALog_ShardInfo::send_response() {
 676   set_req_state_err(s, http_ret);
 677   dump_errno(s);
 678   end_header(s);
 679
 680   encode_json("info", info, s->formatter);
 681   flusher.flush();
 682 }
 683
 684 void RGWOp_DATALog_Lock::execute() {
 685   string shard_id_str, duration_str, locker_id, zone_id;
 686   unsigned shard_id;
 687
 688   http_ret = 0;
 689
 690   shard_id_str = s->info.args.get("id");
 691   duration_str = s->info.args.get("length");
 692   locker_id    = s->info.args.get("locker-id");
 693   zone_id      = s->info.args.get("zone-id");
 694
 695   if (shard_id_str.empty() ||
 696       (duration_str.empty()) ||
 697       locker_id.empty() ||
 698       zone_id.empty()) {
 699     dout(5) << "Error invalid parameter list" << dendl;
 700     http_ret = -EINVAL;
 701     return;
 702   }
 703
 704   string err;
 705   shard_id = (unsigned)strict_strtol(shard_id_str.c_str(), 10, &err);
 706   if (!err.empty()) {
 707     dout(5) << "Error parsing shard_id param " << shard_id_str << dendl;
 708     http_ret = -EINVAL;
 709     return;
 710   }
 711
 712   unsigned dur;
 713   dur = (unsigned)strict_strtol(duration_str.c_str(), 10, &err);
 714   if (!err.empty() || dur <= 0) {
 715     dout(5) << "invalid length param " << duration_str << dendl;
 716     http_ret = -EINVAL;
 717     return;
 718   }
 719   http_ret = store->data_log->lock_exclusive(shard_id, make_timespan(dur), zone_id, locker_id);
 720   if (http_ret == -EBUSY)
 721     http_ret = -ERR_LOCKED;
 722 }
 723
 724 void RGWOp_DATALog_Unlock::execute() {
 725   string shard_id_str, locker_id, zone_id;
 726   unsigned shard_id;
 727
 728   http_ret = 0;
 729
 730   shard_id_str = s->info.args.get("id");
 731   locker_id    = s->info.args.get("locker-id");
 732   zone_id      = s->info.args.get("zone-id");
 733
 734   if (shard_id_str.empty() ||
 735       locker_id.empty() ||
 736       zone_id.empty()) {
 737     dout(5) << "Error invalid parameter list" << dendl;
 738     http_ret = -EINVAL;
 739     return;
 740   }
 741
 742   string err;
 743   shard_id = (unsigned)strict_strtol(shard_id_str.c_str(), 10, &err);
 744   if (!err.empty()) {
 745     dout(5) << "Error parsing shard_id param " << shard_id_str << dendl;
 746     http_ret = -EINVAL;
 747     return;
 748   }
 749
 750   http_ret = store->data_log->unlock(shard_id, zone_id, locker_id);
 751 }
 752
 753 void RGWOp_DATALog_Notify::execute() {
 754   string  source_zone = s->info.args.get("source-zone");
 755   char *data;
 756   int len = 0;
 757 #define LARGE_ENOUGH_BUF (128 * 1024)
 758   int r = rgw_rest_read_all_input(s, &data, &len, LARGE_ENOUGH_BUF);
 759   if (r < 0) {
 760     http_ret = r;
 761     return;
 762   }
 763
 764   ldout(s->cct, 20) << __func__ << "(): read data: " << string(data, len) << dendl;
 765
 766   JSONParser p;
 767   r = p.parse(data, len);
 768   free(data);
 769   if (r < 0) {
 770     ldout(s->cct, 0) << "ERROR: failed to parse JSON" << dendl;
 771     http_ret = r;
 772     return;
 773   }
 774
 775   map<int, set<string> > updated_shards;
 776   try {
 777     decode_json_obj(updated_shards, &p);
 778   } catch (JSONDecoder::err& err) {
 779     ldout(s->cct, 0) << "ERROR: failed to decode JSON" << dendl;
 780     http_ret = -EINVAL;
 781     return;
 782   }
 783
 784   if (store->ctx()->_conf->subsys.should_gather(ceph_subsys_rgw, 20)) {
 785     for (map<int, set<string> >::iterator iter = updated_shards.begin(); iter != updated_shards.end(); ++iter) {
 786       ldout(s->cct, 20) << __func__ << "(): updated shard=" << iter->first << dendl;
 787       set<string>& keys = iter->second;
 788       for (set<string>::iterator kiter = keys.begin(); kiter != keys.end(); ++kiter) {
 789       ldout(s->cct, 20) << __func__ << "(): modified key=" << *kiter << dendl;
 790       }
 791     }
 792   }
 793
 794   store->wakeup_data_sync_shards(source_zone, updated_shards);
 795
 796   http_ret = 0;
 797 }
 798
 799 void RGWOp_DATALog_Delete::execute() {
 800   string   st = s->info.args.get("start-time"),
 801            et = s->info.args.get("end-time"),
 802            start_marker = s->info.args.get("start-marker"),
 803            end_marker = s->info.args.get("end-marker"),
 804            shard = s->info.args.get("id"),
 805            err;
 806   real_time  ut_st,
 807              ut_et;
 808   unsigned shard_id;
 809
 810   http_ret = 0;
 811
 812   shard_id = (unsigned)strict_strtol(shard.c_str(), 10, &err);
 813   if (!err.empty()) {
 814     dout(5) << "Error parsing shard_id " << shard << dendl;
 815     http_ret = -EINVAL;
 816     return;
 817   }
 818   if (et.empty() && end_marker.empty()) { /* bounding end */
 819     http_ret = -EINVAL;
 820     return;
 821   }
 822
 823   if (parse_date_str(st, ut_st) < 0) {
 824     http_ret = -EINVAL;
 825     return;
 826   }
 827
 828   if (parse_date_str(et, ut_et) < 0) {
 829     http_ret = -EINVAL;
 830     return;
 831   }
 832
 833   http_ret = store->data_log->trim_entries(shard_id, ut_st, ut_et, start_marker, end_marker);
 834 }
 835
 836 // not in header to avoid pulling in rgw_sync.h
 837 class RGWOp_MDLog_Status : public RGWRESTOp {
 838   rgw_meta_sync_status status;
 839 public:
 840   int check_caps(RGWUserCaps& caps) override {
 841     return caps.check_cap("mdlog", RGW_CAP_READ);
 842   }
 843   int verify_permission() override {
 844     return check_caps(s->user->caps);
 845   }
 846   void execute() override;
 847   void send_response() override;
 848   const string name() override { return "get_metadata_log_status"; }
 849 };
 850
 851 void RGWOp_MDLog_Status::execute()
 852 {
 853   auto sync = store->get_meta_sync_manager();
 854   if (sync == nullptr) {
 855     ldout(s->cct, 1) << "no sync manager" << dendl;
 856     http_ret = -ENOENT;
 857     return;
 858   }
 859   http_ret = sync->read_sync_status(&status);
 860 }
 861
 862 void RGWOp_MDLog_Status::send_response()
 863 {
 864   set_req_state_err(s, http_ret);
 865   dump_errno(s);
 866   end_header(s);
 867
 868   if (http_ret >= 0) {
 869     encode_json("status", status, s->formatter);
 870   }
 871   flusher.flush();
 872 }
 873
 874 // not in header to avoid pulling in rgw_data_sync.h
 875 class RGWOp_DATALog_Status : public RGWRESTOp {
 876   rgw_data_sync_status status;
 877 public:
 878   int check_caps(RGWUserCaps& caps) override {
 879     return caps.check_cap("datalog", RGW_CAP_READ);
 880   }
 881   int verify_permission() override {
 882     return check_caps(s->user->caps);
 883   }
 884   void execute() override ;
 885   void send_response() override;
 886   const string name() override { return "get_data_changes_log_status"; }
 887 };
 888
 889 void RGWOp_DATALog_Status::execute()
 890 {
 891   const auto source_zone = s->info.args.get("source-zone");
 892   auto sync = store->get_data_sync_manager(source_zone);
 893   if (sync == nullptr) {
 894     ldout(s->cct, 1) << "no sync manager for source-zone " << source_zone << dendl;
 895     http_ret = -ENOENT;
 896     return;
 897   }
 898   http_ret = sync->read_sync_status(&status);
 899 }
 900
 901 void RGWOp_DATALog_Status::send_response()
 902 {
 903   set_req_state_err(s, http_ret);
 904   dump_errno(s);
 905   end_header(s);
 906
 907   if (http_ret >= 0) {
 908     encode_json("status", status, s->formatter);
 909   }
 910   flusher.flush();
 911 }
 912
 913
 914 RGWOp *RGWHandler_Log::op_get() {
 915   bool exists;
 916   string type = s->info.args.get("type", &exists);
 917
 918   if (!exists) {
 919     return NULL;
 920   }
 921
 922   if (type.compare("metadata") == 0) {
 923     if (s->info.args.exists("id")) {
 924       if (s->info.args.exists("info")) {
 925         return new RGWOp_MDLog_ShardInfo;
 926       } else {
 927         return new RGWOp_MDLog_List;
 928       }
 929     } else if (s->info.args.exists("status")) {
 930       return new RGWOp_MDLog_Status;
 931     } else {
 932       return new RGWOp_MDLog_Info;
 933     }
 934   } else if (type.compare("bucket-index") == 0) {
 935     if (s->info.args.exists("info")) {
 936       return new RGWOp_BILog_Info;
 937     } else {
 938       return new RGWOp_BILog_List;
 939     }
 940   } else if (type.compare("data") == 0) {
 941     if (s->info.args.exists("id")) {
 942       if (s->info.args.exists("info")) {
 943         return new RGWOp_DATALog_ShardInfo;
 944       } else {
 945         return new RGWOp_DATALog_List;
 946       }
 947     } else if (s->info.args.exists("status")) {
 948       return new RGWOp_DATALog_Status;
 949     } else {
 950       return new RGWOp_DATALog_Info;
 951     }
 952   }
 953   return NULL;
 954 }
 955
 956 RGWOp *RGWHandler_Log::op_delete() {
 957   bool exists;
 958   string type = s->info.args.get("type", &exists);
 959
 960   if (!exists) {
 961     return NULL;
 962   }
 963
 964   if (type.compare("metadata") == 0)
 965     return new RGWOp_MDLog_Delete;
 966   else if (type.compare("bucket-index") == 0)
 967     return new RGWOp_BILog_Delete;
 968   else if (type.compare("data") == 0)
 969     return new RGWOp_DATALog_Delete;
 970   return NULL;
 971 }
 972
 973 RGWOp *RGWHandler_Log::op_post() {
 974   bool exists;
 975   string type = s->info.args.get("type", &exists);
 976
 977   if (!exists) {
 978     return NULL;
 979   }
 980
 981   if (type.compare("metadata") == 0) {
 982     if (s->info.args.exists("lock"))
 983       return new RGWOp_MDLog_Lock;
 984     else if (s->info.args.exists("unlock"))
 985       return new RGWOp_MDLog_Unlock;
 986     else if (s->info.args.exists("notify"))
 987       return new RGWOp_MDLog_Notify;
 988   } else if (type.compare("data") == 0) {
 989     if (s->info.args.exists("lock"))
 990       return new RGWOp_DATALog_Lock;
 991     else if (s->info.args.exists("unlock"))
 992       return new RGWOp_DATALog_Unlock;
 993     else if (s->info.args.exists("notify"))
 994       return new RGWOp_DATALog_Notify;
 995   }
 996   return NULL;
 997 }
 998