ceph/src/rgw/rgw_object_expirer_core.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 #ifndef CEPH_OBJEXP_H
   5 #define CEPH_OBJEXP_H
   6
   7 #include <atomic>
   8 #include <string>
   9 #include <cerrno>
  10 #include <sstream>
  11 #include <iostream>
  12
  13 #include "auth/Crypto.h"
  14
  15 #include "common/armor.h"
  16 #include "common/ceph_json.h"
  17 #include "common/config.h"
  18 #include "common/ceph_argparse.h"
  19 #include "common/Formatter.h"
  20 #include "common/errno.h"
  21
  22 #include "common/ceph_mutex.h"
  23 #include "common/Cond.h"
  24 #include "common/Thread.h"
  25
  26 #include "global/global_init.h"
  27
  28 #include "include/common_fwd.h"
  29 #include "include/utime.h"
  30 #include "include/str_list.h"
  31
  32 #include "rgw_sal.h"
  33 #include "rgw_sal_rados.h"
  34
  35 class RGWSI_RADOS;
  36 class RGWSI_Zone;
  37 class RGWBucketInfo;
  38 class cls_timeindex_entry;
  39
  40 class RGWObjExpStore {
  41   CephContext *cct;
  42   RGWSI_RADOS *rados_svc;
  43   RGWSI_Zone *zone_svc;
  44 public:
  45   RGWObjExpStore(CephContext *_cct, RGWSI_RADOS *_rados_svc, RGWSI_Zone *_zone_svc) : cct(_cct),
  46                                                                                       rados_svc(_rados_svc),
  47                                                                                       zone_svc(_zone_svc) {}
  48
  49   int objexp_hint_add(const DoutPrefixProvider *dpp,
  50                       const ceph::real_time& delete_at,
  51                       const string& tenant_name,
  52                       const string& bucket_name,
  53                       const string& bucket_id,
  54                       const rgw_obj_index_key& obj_key);
  55
  56   int objexp_hint_list(const DoutPrefixProvider *dpp,
  57                        const string& oid,
  58                        const ceph::real_time& start_time,
  59                        const ceph::real_time& end_time,
  60                        const int max_entries,
  61                        const string& marker,
  62                        list<cls_timeindex_entry>& entries, /* out */
  63                        string *out_marker,                 /* out */
  64                        bool *truncated);                   /* out */
  65
  66   int objexp_hint_trim(const DoutPrefixProvider *dpp,
  67                        const string& oid,
  68                        const ceph::real_time& start_time,
  69                        const ceph::real_time& end_time,
  70                        const string& from_marker,
  71                        const string& to_marker);
  72 };
  73
  74 class RGWObjectExpirer {
  75 protected:
  76   rgw::sal::RGWRadosStore *store;
  77   RGWObjExpStore exp_store;
  78
  79   int init_bucket_info(const std::string& tenant_name,
  80                        const std::string& bucket_name,
  81                        const std::string& bucket_id,
  82                        RGWBucketInfo& bucket_info);
  83
  84   class OEWorker : public Thread, public DoutPrefixProvider {
  85     CephContext *cct;
  86     RGWObjectExpirer *oe;
  87     ceph::mutex lock = ceph::make_mutex("OEWorker");
  88     ceph::condition_variable cond;
  89
  90   public:
  91     OEWorker(CephContext * const cct,
  92              RGWObjectExpirer * const oe)
  93       : cct(cct),
  94         oe(oe) {
  95     }
  96
  97     void *entry() override;
  98     void stop();
  99
 100     CephContext *get_cct() const override;
 101     unsigned get_subsys() const;
 102     std::ostream& gen_prefix(std::ostream& out) const;
 103   };
 104
 105   OEWorker *worker{nullptr};
 106   std::atomic<bool> down_flag = { false };
 107
 108 public:
 109   explicit RGWObjectExpirer(rgw::sal::RGWRadosStore *_store)
 110     : store(_store),
 111       exp_store(_store->getRados()->ctx(), _store->svc()->rados, _store->svc()->zone),
 112       worker(NULL) {
 113   }
 114   ~RGWObjectExpirer() {
 115     stop_processor();
 116   }
 117
 118   int hint_add(const DoutPrefixProvider *dpp,
 119                const ceph::real_time& delete_at,
 120                const string& tenant_name,
 121                const string& bucket_name,
 122                const string& bucket_id,
 123                const rgw_obj_index_key& obj_key) {
 124     return exp_store.objexp_hint_add(dpp, delete_at, tenant_name, bucket_name,
 125                                      bucket_id, obj_key);
 126   }
 127
 128   int garbage_single_object(const DoutPrefixProvider *dpp, objexp_hint_entry& hint);
 129
 130   void garbage_chunk(const DoutPrefixProvider *dpp,
 131                      std::list<cls_timeindex_entry>& entries, /* in  */
 132                      bool& need_trim);                        /* out */
 133
 134   void trim_chunk(const DoutPrefixProvider *dpp,
 135                   const std::string& shard,
 136                   const utime_t& from,
 137                   const utime_t& to,
 138                   const string& from_marker,
 139                   const string& to_marker);
 140
 141   bool process_single_shard(const DoutPrefixProvider *dpp,
 142                             const std::string& shard,
 143                             const utime_t& last_run,
 144                             const utime_t& round_start);
 145
 146   bool inspect_all_shards(const DoutPrefixProvider *dpp,
 147                           const utime_t& last_run,
 148                           const utime_t& round_start);
 149
 150   bool going_down();
 151   void start_processor();
 152   void stop_processor();
 153 };
 154 #endif /* CEPH_OBJEXP_H */