path: root/src/rgw/rgw_lc.h

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab ft=cpp

#pragma once

#include <map>
#include <array>
#include <string>
#include <iostream>

#include "common/debug.h"

#include "include/types.h"
#include "include/rados/librados.hpp"
#include "common/ceph_mutex.h"
#include "common/Cond.h"
#include "common/iso_8601.h"
#include "common/Thread.h"
#include "rgw_common.h"
#include "cls/rgw/cls_rgw_types.h"
#include "rgw_tag.h"
#include "rgw_sal.h"

#include <atomic>
#include <tuple>

#define HASH_PRIME 7877
#define MAX_ID_LEN 255
static std::string lc_oid_prefix = "lc";
static std::string lc_index_lock_name = "lc_process";

extern const char* LC_STATUS[];

typedef enum {
  lc_uninitial = 0,
  lc_processing,
  lc_failed,
  lc_complete,
} LC_BUCKET_STATUS;

class LCExpiration
{
protected:
  std::string days;
  //At present only current object has expiration date
  std::string date;
public:
  LCExpiration() {}
  LCExpiration(const std::string& _days, const std::string& _date) : days(_days), date(_date) {}

  void encode(bufferlist& bl) const {
    ENCODE_START(3, 2, bl);
    encode(days, bl);
    encode(date, bl);
    ENCODE_FINISH(bl);
  }
  void decode(bufferlist::const_iterator& bl) {
    DECODE_START_LEGACY_COMPAT_LEN(3, 2, 2, bl);
    decode(days, bl);
    if (struct_v >= 3) {
      decode(date, bl);
    }
    DECODE_FINISH(bl);
  }
  void dump(Formatter *f) const;
//  static void generate_test_instances(list<ACLOwner*>& o);
  void set_days(const std::string& _days) { days = _days; }
  std::string get_days_str() const {
    return days;
  }
  int get_days() const {return atoi(days.c_str()); }
  bool has_days() const {
    return !days.empty();
  }
  void set_date(const std::string& _date) { date = _date; }
  std::string get_date() const {
    return date;
  }
  bool has_date() const {
    return !date.empty();
  }
  bool empty() const {
    return days.empty() && date.empty();
  }
  bool valid() const {
    if (!days.empty() && !date.empty()) {
      return false;
    } else if (!days.empty() && get_days() <= 0) {
      return false;
    }
    //We've checked date in xml parsing
    return true;
  }
};
WRITE_CLASS_ENCODER(LCExpiration)

class LCTransition
{
protected:
  std::string days;
  std::string date;
  std::string storage_class;

public:
  int get_days() const {
    return atoi(days.c_str());
  }

  std::string get_date() const {
    return date;
  }

  std::string get_storage_class() const {
    return storage_class;
  }

  bool has_days() const {
    return !days.empty();
  }

  bool has_date() const {
    return !date.empty();
  }

  bool empty() const {
    return days.empty() && date.empty();
  }

  bool valid() const {
    if (!days.empty() && !date.empty()) {
      return false;
    } else if (!days.empty() && get_days() < 0) {
      return false;
    }
    //We've checked date in xml parsing
    return true;
  }

  void encode(bufferlist& bl) const {
    ENCODE_START(1, 1, bl);
    encode(days, bl);
    encode(date, bl);
    encode(storage_class, bl);
    ENCODE_FINISH(bl);
  }

  void decode(bufferlist::const_iterator& bl) {
    DECODE_START(1, bl);
    decode(days, bl);
    decode(date, bl);
    decode(storage_class, bl);
    DECODE_FINISH(bl);
  }
  void dump(Formatter *f) const {  
    f->dump_string("days", days);
    f->dump_string("date", date);
    f->dump_string("storage_class", storage_class);
  }
};
WRITE_CLASS_ENCODER(LCTransition)

enum class LCFlagType : uint16_t
{
  none = 0,
  ArchiveZone,
};

class LCFlag {
public:
  LCFlagType bit;
  const char* name;

  constexpr LCFlag(LCFlagType ord, const char* name) : bit(ord), name(name)
    {}
};

class LCFilter
{
 public:

  static constexpr uint32_t make_flag(LCFlagType type) {
    switch (type) {
    case LCFlagType::none:
      return 0;
      break;
    default:
      return 1 << (uint32_t(type) - 1);
    }
   }

  static constexpr std::array<LCFlag, 2> filter_flags =
  {
    LCFlag(LCFlagType::none, "none"),
    LCFlag(LCFlagType::ArchiveZone, "ArchiveZone"),
  };

protected:
  std::string prefix;
  RGWObjTags obj_tags;
  uint32_t flags;

public:

  LCFilter() : flags(make_flag(LCFlagType::none))
    {}

  const std::string& get_prefix() const {
    return prefix;
  }

  const RGWObjTags& get_tags() const {
    return obj_tags;
  }

  const uint32_t get_flags() const {
    return flags;
  }

  bool empty() const {
    return !(has_prefix() || has_tags() || has_flags());
  }

  // Determine if we need AND tag when creating xml
  bool has_multi_condition() const {
    if (obj_tags.count() + int(has_prefix()) + int(has_flags()) > 1) // Prefix is a member of Filter
      return true;
    return false;
  }

  bool has_prefix() const {
    return !prefix.empty();
  }

  bool has_tags() const {
    return !obj_tags.empty();
  }

  bool has_flags() const {
    return !(flags == uint32_t(LCFlagType::none));
  }

  bool have_flag(LCFlagType flag) const {
    return flags & make_flag(flag);
  }

  void encode(bufferlist& bl) const {
    ENCODE_START(3, 1, bl);
    encode(prefix, bl);
    encode(obj_tags, bl);
    encode(flags, bl);
    ENCODE_FINISH(bl);
  }
  void decode(bufferlist::const_iterator& bl) {
    DECODE_START(3, bl);
    decode(prefix, bl);
    if (struct_v >= 2) {
      decode(obj_tags, bl);
      if (struct_v >= 3) {
	decode(flags, bl);
      }
    }
    DECODE_FINISH(bl);
  }
  void dump(Formatter *f) const;
};
WRITE_CLASS_ENCODER(LCFilter)

class LCRule
{
protected:
  std::string id;
  std::string prefix;
  std::string status;
  LCExpiration expiration;
  LCExpiration noncur_expiration;
  LCExpiration mp_expiration;
  LCFilter filter;
  std::map<std::string, LCTransition> transitions;
  std::map<std::string, LCTransition> noncur_transitions;
  bool dm_expiration = false;

public:

  LCRule(){};
  virtual ~LCRule() {}

  const std::string& get_id() const {
      return id;
  }

  const std::string& get_status() const {
      return status;
  }

  bool is_enabled() const {
    return status == "Enabled";
  }

  void set_enabled(bool flag) {
    status = (flag ? "Enabled" : "Disabled");
  }

  const std::string& get_prefix() const {
      return prefix;
  }

  const LCFilter& get_filter() const {
    return filter;
  }

  const LCExpiration& get_expiration() const {
    return expiration;
  }

  const LCExpiration& get_noncur_expiration() const {
    return noncur_expiration;
  }

  const LCExpiration& get_mp_expiration() const {
    return mp_expiration;
  }

  bool get_dm_expiration() const {
    return dm_expiration;
  }

  const std::map<std::string, LCTransition>& get_transitions() const {
    return transitions;
  }

  const std::map<std::string, LCTransition>& get_noncur_transitions() const {
    return noncur_transitions;
  }

  void set_id(const std::string& _id) {
    id = _id;
  }

  void set_prefix(const std::string& _prefix) {
    prefix = _prefix;
  }

  void set_status(const std::string& _status) {
    status = _status;
  }

  void set_expiration(const LCExpiration& _expiration) {
    expiration = _expiration;
  }

  void set_noncur_expiration(const LCExpiration& _noncur_expiration) {
    noncur_expiration = _noncur_expiration;
  }

  void set_mp_expiration(const LCExpiration& _mp_expiration) {
    mp_expiration = _mp_expiration;
  }

  void set_dm_expiration(bool _dm_expiration) {
    dm_expiration = _dm_expiration;
  }

  bool add_transition(const LCTransition& _transition) {
    auto ret = transitions.emplace(_transition.get_storage_class(), _transition);
    return ret.second;
  }

  bool add_noncur_transition(const LCTransition& _noncur_transition) {
    auto ret = noncur_transitions.emplace(_noncur_transition.get_storage_class(), _noncur_transition);
    return ret.second;
  }

  bool valid() const;
  
  void encode(bufferlist& bl) const {
     ENCODE_START(6, 1, bl);
     encode(id, bl);
     encode(prefix, bl);
     encode(status, bl);
     encode(expiration, bl);
     encode(noncur_expiration, bl);
     encode(mp_expiration, bl);
     encode(dm_expiration, bl);
     encode(filter, bl);
     encode(transitions, bl);
     encode(noncur_transitions, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::const_iterator& bl) {
     DECODE_START_LEGACY_COMPAT_LEN(6, 1, 1, bl);
     decode(id, bl);
     decode(prefix, bl);
     decode(status, bl);
     decode(expiration, bl);
     if (struct_v >=2) {
       decode(noncur_expiration, bl);
     }
     if (struct_v >= 3) {
       decode(mp_expiration, bl);
     }
     if (struct_v >= 4) {
        decode(dm_expiration, bl);
     }
     if (struct_v >= 5) {
       decode(filter, bl);
     }
     if (struct_v >= 6) {
       decode(transitions, bl);
       decode(noncur_transitions, bl);
     }
     DECODE_FINISH(bl);
   }
  void dump(Formatter *f) const;

  void init_simple_days_rule(std::string_view _id, std::string_view _prefix, int num_days);
};
WRITE_CLASS_ENCODER(LCRule)

struct transition_action
{
  int days;
  boost::optional<ceph::real_time> date;
  std::string storage_class;
  transition_action() : days(0) {}
  void dump(Formatter *f) const {
    if (!date) {
      f->dump_int("days", days);
    } else {
      utime_t ut(*date);
      f->dump_stream("date") << ut;
    }
  }
};

/* XXX why not LCRule? */
struct lc_op
{
  std::string id;
  bool status{false};
  bool dm_expiration{false};
  int expiration{0};
  int noncur_expiration{0};
  int mp_expiration{0};
  boost::optional<ceph::real_time> expiration_date;
  boost::optional<RGWObjTags> obj_tags;
  std::map<std::string, transition_action> transitions;
  std::map<std::string, transition_action> noncur_transitions;
  uint32_t rule_flags;

  /* ctors are nice */
  lc_op() = delete;

  lc_op(const std::string id) : id(id)
    {}

  void dump(Formatter *f) const;
};

class RGWLifecycleConfiguration
{
protected:
  CephContext *cct;
  std::multimap<std::string, lc_op> prefix_map;
  std::multimap<std::string, LCRule> rule_map;
  bool _add_rule(const LCRule& rule);
  bool has_same_action(const lc_op& first, const lc_op& second);
public:
  explicit RGWLifecycleConfiguration(CephContext *_cct) : cct(_cct) {}
  RGWLifecycleConfiguration() : cct(NULL) {}

  void set_ctx(CephContext *ctx) {
    cct = ctx;
  }

  virtual ~RGWLifecycleConfiguration() {}

//  int get_perm(std::string& id, int perm_mask);
//  int get_group_perm(ACLGroupTypeEnum group, int perm_mask);
  void encode(bufferlist& bl) const {
    ENCODE_START(1, 1, bl);
    encode(rule_map, bl);
    ENCODE_FINISH(bl);
  }
  void decode(bufferlist::const_iterator& bl) {
    DECODE_START_LEGACY_COMPAT_LEN(1, 1, 1, bl);
    decode(rule_map, bl);
    std::multimap<std::string, LCRule>::iterator iter;
    for (iter = rule_map.begin(); iter != rule_map.end(); ++iter) {
      LCRule& rule = iter->second;
      _add_rule(rule);
    }
    DECODE_FINISH(bl);
  }
  void dump(Formatter *f) const;
  static void generate_test_instances(std::list<RGWLifecycleConfiguration*>& o);

  void add_rule(const LCRule& rule);

  int check_and_add_rule(const LCRule& rule);

  bool valid();

  std::multimap<std::string, LCRule>& get_rule_map() { return rule_map; }
  std::multimap<std::string, lc_op>& get_prefix_map() { return prefix_map; }
/*
  void create_default(std::string id, std::string name) {
    ACLGrant grant;
    grant.set_canon(id, name, RGW_PERM_FULL_CONTROL);
    add_grant(&grant);
  }
*/
};
WRITE_CLASS_ENCODER(RGWLifecycleConfiguration)

class RGWLC : public DoutPrefixProvider {
  CephContext *cct;
  rgw::sal::Driver* driver;
  std::unique_ptr<rgw::sal::Lifecycle> sal_lc;
  int max_objs{0};
  std::string *obj_names{nullptr};
  std::atomic<bool> down_flag = { false };
  std::string cookie;

public:

  class WorkPool;

  class LCWorker : public Thread
  {
    const DoutPrefixProvider *dpp;
    CephContext *cct;
    RGWLC *lc;
    int ix;
    std::mutex lock;
    std::condition_variable cond;
    WorkPool* workpool{nullptr};
    /* save the target bucket names created as part of object transition
     * to cloud. This list is maintained for the duration of each RGWLC::process()
     * post which it is discarded. */
    std::set<std::string> cloud_targets;

  public:

    using lock_guard = std::lock_guard<std::mutex>;
    using unique_lock = std::unique_lock<std::mutex>;

    LCWorker(const DoutPrefixProvider* dpp, CephContext *_cct, RGWLC *_lc,
	     int ix);
    RGWLC* get_lc() { return lc; }

    std::string thr_name() {
      return std::string{"lc_thrd: "} + std::to_string(ix);
    }

    void *entry() override;
    void stop();
    bool should_work(utime_t& now);
    int schedule_next_start_time(utime_t& start, utime_t& now);
    std::set<std::string>& get_cloud_targets() { return cloud_targets; }
    virtual ~LCWorker() override;

    friend class RGWRados;
    friend class RGWLC;
    friend class WorkQ;
  }; /* LCWorker */

  friend class RGWRados;

  std::vector<std::unique_ptr<RGWLC::LCWorker>> workers;

  RGWLC() : cct(nullptr), driver(nullptr) {}
  virtual ~RGWLC() override;

  void initialize(CephContext *_cct, rgw::sal::Driver* _driver);
  void finalize();

  int process(LCWorker* worker,
	      const std::unique_ptr<rgw::sal::Bucket>& optional_bucket,
	      bool once);
  int advance_head(const std::string& lc_shard,
		   rgw::sal::Lifecycle::LCHead& head,
		   rgw::sal::Lifecycle::LCEntry& entry,
		   time_t start_date);
  int process(int index, int max_lock_secs, LCWorker* worker, bool once);
  int process_bucket(int index, int max_lock_secs, LCWorker* worker,
		     const std::string& bucket_entry_marker, bool once);
  bool expired_session(time_t started);
  time_t thread_stop_at();
  int list_lc_progress(std::string& marker, uint32_t max_entries,
		       std::vector<std::unique_ptr<rgw::sal::Lifecycle::LCEntry>>&,
		       int& index);
  int bucket_lc_process(std::string& shard_id, LCWorker* worker, time_t stop_at,
			bool once);
  int bucket_lc_post(int index, int max_lock_sec,
		     rgw::sal::Lifecycle::LCEntry& entry, int& result, LCWorker* worker);
  bool going_down();
  void start_processor();
  void stop_processor();
  int set_bucket_config(rgw::sal::Bucket* bucket,
                        const rgw::sal::Attrs& bucket_attrs,
                        RGWLifecycleConfiguration *config);
  int remove_bucket_config(rgw::sal::Bucket* bucket,
                           const rgw::sal::Attrs& bucket_attrs,
			   bool merge_attrs = true);

  CephContext *get_cct() const override { return cct; }
  rgw::sal::Lifecycle* get_lc() const { return sal_lc.get(); }
  unsigned get_subsys() const;
  std::ostream& gen_prefix(std::ostream& out) const;

  private:

  int handle_multipart_expiration(rgw::sal::Bucket* target,
				  const std::multimap<std::string, lc_op>& prefix_map,
				  LCWorker* worker, time_t stop_at, bool once);
};

namespace rgw::lc {

int fix_lc_shard_entry(const DoutPrefixProvider *dpp,
                       rgw::sal::Driver* driver,
		       rgw::sal::Lifecycle* sal_lc,
		       rgw::sal::Bucket* bucket);

std::string s3_expiration_header(
  DoutPrefixProvider* dpp,
  const rgw_obj_key& obj_key,
  const RGWObjTags& obj_tagset,
  const ceph::real_time& mtime,
  const std::map<std::string, buffer::list>& bucket_attrs);

bool s3_multipart_abort_header(
  DoutPrefixProvider* dpp,
  const rgw_obj_key& obj_key,
  const ceph::real_time& mtime,
  const std::map<std::string, buffer::list>& bucket_attrs,
  ceph::real_time& abort_date,
  std::string& rule_id);

} // namespace rgw::lc