+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-
-#include <errno.h>
-#include <iostream>
-#include <sstream>
-#include <string>
-
-using namespace std;
-
-#include "auth/Crypto.h"
-
-#include "common/armor.h"
-#include "common/ceph_json.h"
-#include "common/config.h"
-#include "common/ceph_argparse.h"
-#include "common/Formatter.h"
-#include "common/errno.h"
-
-#include "global/global_init.h"
-
-#include "include/utime.h"
-#include "include/str_list.h"
-
-#include "rgw_user.h"
-#include "rgw_bucket.h"
-#include "rgw_rados.h"
-#include "rgw_acl.h"
-#include "rgw_acl_s3.h"
-#include "rgw_log.h"
-#include "rgw_formats.h"
-#include "rgw_usage.h"
-#include "rgw_replica_log.h"
-#include "rgw_object_expirer_core.h"
-
-#include "cls/lock/cls_lock_client.h"
-
-#define dout_context g_ceph_context
-#define dout_subsys ceph_subsys_rgw
-
-static string objexp_lock_name = "gc_process";
-
-int RGWObjectExpirer::init_bucket_info(const string& tenant_name,
- const string& bucket_name,
- const string& bucket_id,
- RGWBucketInfo& bucket_info)
-{
- RGWObjectCtx obj_ctx(store);
-
- /*
- * XXX Here's where it gets tricky. We went to all the trouble of
- * punching the tenant through the objexp_hint_entry, but now we
- * find that our instances do not actually have tenants. They are
- * unique thanks to IDs. So the tenant string is not needed...
- */
- const string bucket_instance_id = bucket_name + ":" + bucket_id;
- int ret = store->get_bucket_instance_info(obj_ctx, bucket_instance_id,
- bucket_info, NULL, NULL);
- return ret;
-}
-
-int RGWObjectExpirer::garbage_single_object(objexp_hint_entry& hint)
-{
- RGWBucketInfo bucket_info;
-
- int ret = init_bucket_info(hint.tenant, hint.bucket_name,
- hint.bucket_id, bucket_info);
- if (-ENOENT == ret) {
- ldout(store->ctx(), 15) << "NOTICE: cannot find bucket = " \
- << hint.bucket_name << ". The object must be already removed" << dendl;
- return -ERR_PRECONDITION_FAILED;
- } else if (ret < 0) {
- ldout(store->ctx(), 1) << "ERROR: could not init bucket = " \
- << hint.bucket_name << "due to ret = " << ret << dendl;
- return ret;
- }
-
- RGWObjectCtx rctx(store);
-
- rgw_obj_key key = hint.obj_key;
- if (key.instance.empty()) {
- key.instance = "null";
- }
-
- rgw_obj obj(bucket_info.bucket, key);
- store->set_atomic(&rctx, obj);
- ret = store->delete_obj(rctx, bucket_info, obj,
- bucket_info.versioning_status(), 0, hint.exp_time);
-
- return ret;
-}
-
-void RGWObjectExpirer::garbage_chunk(list<cls_timeindex_entry>& entries, /* in */
- bool& need_trim) /* out */
-{
- need_trim = false;
-
- for (list<cls_timeindex_entry>::iterator iter = entries.begin();
- iter != entries.end();
- ++iter)
- {
- objexp_hint_entry hint;
- ldout(store->ctx(), 15) << "got removal hint for: " << iter->key_ts.sec() \
- << " - " << iter->key_ext << dendl;
-
- int ret = store->objexp_hint_parse(*iter, hint);
- if (ret < 0) {
- ldout(store->ctx(), 1) << "cannot parse removal hint for " << hint.obj_key << dendl;
- continue;
- }
-
- /* PRECOND_FAILED simply means that our hint is not valid.
- * We can silently ignore that and move forward. */
- ret = garbage_single_object(hint);
- if (ret == -ERR_PRECONDITION_FAILED) {
- ldout(store->ctx(), 15) << "not actual hint for object: " << hint.obj_key << dendl;
- } else if (ret < 0) {
- ldout(store->ctx(), 1) << "cannot remove expired object: " << hint.obj_key << dendl;
- }
-
- need_trim = true;
- }
-
- return;
-}
-
-void RGWObjectExpirer::trim_chunk(const string& shard,
- const utime_t& from,
- const utime_t& to,
- const string& from_marker,
- const string& to_marker)
-{
- ldout(store->ctx(), 20) << "trying to trim removal hints to=" << to
- << ", to_marker=" << to_marker << dendl;
-
- real_time rt_from = from.to_real_time();
- real_time rt_to = to.to_real_time();
-
- int ret = store->objexp_hint_trim(shard, rt_from, rt_to,
- from_marker, to_marker);
- if (ret < 0) {
- ldout(store->ctx(), 0) << "ERROR during trim: " << ret << dendl;
- }
-
- return;
-}
-
-bool RGWObjectExpirer::process_single_shard(const string& shard,
- const utime_t& last_run,
- const utime_t& round_start)
-{
- string marker;
- string out_marker;
- bool truncated = false;
- bool done = true;
-
- CephContext *cct = store->ctx();
- int num_entries = cct->_conf->rgw_objexp_chunk_size;
-
- int max_secs = cct->_conf->rgw_objexp_gc_interval;
- utime_t end = ceph_clock_now();
- end += max_secs;
-
- rados::cls::lock::Lock l(objexp_lock_name);
-
- utime_t time(max_secs, 0);
- l.set_duration(time);
-
- int ret = l.lock_exclusive(&store->objexp_pool_ctx, shard);
- if (ret == -EBUSY) { /* already locked by another processor */
- dout(5) << __func__ << "(): failed to acquire lock on " << shard << dendl;
- return false;
- }
-
- do {
- real_time rt_last = last_run.to_real_time();
- real_time rt_start = round_start.to_real_time();
-
- list<cls_timeindex_entry> entries;
- ret = store->objexp_hint_list(shard, rt_last, rt_start,
- num_entries, marker, entries,
- &out_marker, &truncated);
- if (ret < 0) {
- ldout(cct, 10) << "cannot get removal hints from shard: " << shard
- << dendl;
- continue;
- }
-
- bool need_trim;
- garbage_chunk(entries, need_trim);
-
- if (need_trim) {
- trim_chunk(shard, last_run, round_start, marker, out_marker);
- }
-
- utime_t now = ceph_clock_now();
- if (now >= end) {
- done = false;
- break;
- }
-
- marker = out_marker;
- } while (truncated);
-
- l.unlock(&store->objexp_pool_ctx, shard);
- return done;
-}
-
-/* Returns true if all shards have been processed successfully. */
-bool RGWObjectExpirer::inspect_all_shards(const utime_t& last_run,
- const utime_t& round_start)
-{
- CephContext * const cct = store->ctx();
- int num_shards = cct->_conf->rgw_objexp_hints_num_shards;
- bool all_done = true;
-
- for (int i = 0; i < num_shards; i++) {
- string shard;
- store->objexp_get_shard(i, shard);
-
- ldout(store->ctx(), 20) << "proceeding shard = " << shard << dendl;
-
- if (! process_single_shard(shard, last_run, round_start)) {
- all_done = false;
- }
- }
-
- return all_done;
-}
-
-bool RGWObjectExpirer::going_down()
-{
- return down_flag;
-}
-
-void RGWObjectExpirer::start_processor()
-{
- worker = new OEWorker(store->ctx(), this);
- worker->create("rgw_obj_expirer");
-}
-
-void RGWObjectExpirer::stop_processor()
-{
- down_flag = true;
- if (worker) {
- worker->stop();
- worker->join();
- }
- delete worker;
- worker = NULL;
-}
-
-void *RGWObjectExpirer::OEWorker::entry() {
- utime_t last_run;
- do {
- utime_t start = ceph_clock_now();
- ldout(cct, 2) << "object expiration: start" << dendl;
- if (oe->inspect_all_shards(last_run, start)) {
- /* All shards have been processed properly. Next time we can start
- * from this moment. */
- last_run = start;
- }
- ldout(cct, 2) << "object expiration: stop" << dendl;
-
-
- if (oe->going_down())
- break;
-
- utime_t end = ceph_clock_now();
- end -= start;
- int secs = cct->_conf->rgw_objexp_gc_interval;
-
- if (secs <= end.sec())
- continue; // next round
-
- secs -= end.sec();
-
- lock.Lock();
- cond.WaitInterval(lock, utime_t(secs, 0));
- lock.Unlock();
- } while (!oe->going_down());
-
- return NULL;
-}
-
-void RGWObjectExpirer::OEWorker::stop()
-{
- Mutex::Locker l(lock);
- cond.Signal();
-}
-
Code Review / stor4nfv.git / blobdiff