ceph/src/rgw/rgw_log.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_RGW_LOG_H
   5 #define CEPH_RGW_LOG_H
   6
   7 #include <boost/container/flat_map.hpp>
   8 #include "rgw_common.h"
   9 #include "common/OutputDataSocket.h"
  10 #include <vector>
  11 #include <fstream>
  12
  13 #define dout_subsys ceph_subsys_rgw
  14
  15 namespace rgw { namespace sal {
  16   class Store;
  17 } }
  18
  19 struct rgw_log_entry {
  20
  21   using headers_map = boost::container::flat_map<std::string, std::string>;
  22   using Clock = req_state::Clock;
  23
  24   rgw_user object_owner;
  25   rgw_user bucket_owner;
  26   std::string bucket;
  27   Clock::time_point time;
  28   std::string remote_addr;
  29   std::string user;
  30   rgw_obj_key obj;
  31   std::string op;
  32   std::string uri;
  33   std::string http_status;
  34   std::string error_code;
  35   uint64_t bytes_sent = 0;
  36   uint64_t bytes_received = 0;
  37   uint64_t obj_size = 0;
  38   Clock::duration total_time{};
  39   std::string user_agent;
  40   std::string referrer;
  41   std::string bucket_id;
  42   headers_map x_headers;
  43   std::string trans_id;
  44   std::vector<std::string> token_claims;
  45   uint32_t identity_type;
  46   std::string access_key_id;
  47   std::string subuser;
  48   bool temp_url {false};
  49
  50   void encode(bufferlist &bl) const {
  51     ENCODE_START(13, 5, bl);
  52     encode(object_owner.id, bl);
  53     encode(bucket_owner.id, bl);
  54     encode(bucket, bl);
  55     encode(time, bl);
  56     encode(remote_addr, bl);
  57     encode(user, bl);
  58     encode(obj.name, bl);
  59     encode(op, bl);
  60     encode(uri, bl);
  61     encode(http_status, bl);
  62     encode(error_code, bl);
  63     encode(bytes_sent, bl);
  64     encode(obj_size, bl);
  65     encode(total_time, bl);
  66     encode(user_agent, bl);
  67     encode(referrer, bl);
  68     encode(bytes_received, bl);
  69     encode(bucket_id, bl);
  70     encode(obj, bl);
  71     encode(object_owner, bl);
  72     encode(bucket_owner, bl);
  73     encode(x_headers, bl);
  74     encode(trans_id, bl);
  75     encode(token_claims, bl);
  76     encode(identity_type,bl);
  77     encode(access_key_id, bl);
  78     encode(subuser, bl);
  79     encode(temp_url, bl);
  80     ENCODE_FINISH(bl);
  81   }
  82   void decode(bufferlist::const_iterator &p) {
  83     DECODE_START_LEGACY_COMPAT_LEN(13, 5, 5, p);
  84     decode(object_owner.id, p);
  85     if (struct_v > 3)
  86       decode(bucket_owner.id, p);
  87     decode(bucket, p);
  88     decode(time, p);
  89     decode(remote_addr, p);
  90     decode(user, p);
  91     decode(obj.name, p);
  92     decode(op, p);
  93     decode(uri, p);
  94     decode(http_status, p);
  95     decode(error_code, p);
  96     decode(bytes_sent, p);
  97     decode(obj_size, p);
  98     decode(total_time, p);
  99     decode(user_agent, p);
 100     decode(referrer, p);
 101     if (struct_v >= 2)
 102       decode(bytes_received, p);
 103     else
 104       bytes_received = 0;
 105
 106     if (struct_v >= 3) {
 107       if (struct_v <= 5) {
 108         uint64_t id;
 109         decode(id, p);
 110         char buf[32];
 111         snprintf(buf, sizeof(buf), "%" PRIu64, id);
 112         bucket_id = buf;
 113       } else {
 114         decode(bucket_id, p);
 115       }
 116     } else {
 117       bucket_id = "";
 118     }
 119     if (struct_v >= 7) {
 120       decode(obj, p);
 121     }
 122     if (struct_v >= 8) {
 123       decode(object_owner, p);
 124       decode(bucket_owner, p);
 125     }
 126     if (struct_v >= 9) {
 127       decode(x_headers, p);
 128     }
 129     if (struct_v >= 10) {
 130       decode(trans_id, p);
 131     }
 132     if (struct_v >= 11) {
 133       decode(token_claims, p);
 134     }
 135     if (struct_v >= 12) {
 136       decode(identity_type, p);
 137     }
 138     if (struct_v >= 13) {
 139       decode(access_key_id, p);
 140       decode(subuser, p);
 141       decode(temp_url, p);
 142     }
 143     DECODE_FINISH(p);
 144   }
 145   void dump(ceph::Formatter *f) const;
 146   static void generate_test_instances(std::list<rgw_log_entry*>& o);
 147 };
 148 WRITE_CLASS_ENCODER(rgw_log_entry)
 149
 150 class OpsLogSink {
 151 public:
 152   virtual int log(struct req_state* s, struct rgw_log_entry& entry) = 0;
 153   virtual ~OpsLogSink() = default;
 154 };
 155
 156 class OpsLogManifold: public OpsLogSink {
 157   std::vector<OpsLogSink*> sinks;
 158 public:
 159   ~OpsLogManifold() override;
 160   void add_sink(OpsLogSink* sink);
 161   int log(struct req_state* s, struct rgw_log_entry& entry) override;
 162 };
 163
 164 class JsonOpsLogSink : public OpsLogSink {
 165   ceph::Formatter *formatter;
 166   ceph::mutex lock = ceph::make_mutex("JsonOpsLogSink");
 167
 168   void formatter_to_bl(bufferlist& bl);
 169 protected:
 170   virtual int log_json(struct req_state* s, bufferlist& bl) = 0;
 171 public:
 172   JsonOpsLogSink();
 173   ~JsonOpsLogSink() override;
 174   int log(struct req_state* s, struct rgw_log_entry& entry) override;
 175 };
 176
 177 class OpsLogFile : public JsonOpsLogSink, public Thread, public DoutPrefixProvider {
 178   CephContext* cct;
 179   ceph::mutex mutex = ceph::make_mutex("OpsLogFile");
 180   std::vector<bufferlist> log_buffer;
 181   std::vector<bufferlist> flush_buffer;
 182   ceph::condition_variable cond;
 183   std::ofstream file;
 184   bool stopped;
 185   uint64_t data_size;
 186   uint64_t max_data_size;
 187   std::string path;
 188   std::atomic_bool need_reopen;
 189
 190   void flush();
 191 protected:
 192   int log_json(struct req_state* s, bufferlist& bl) override;
 193   void *entry() override;
 194 public:
 195   OpsLogFile(CephContext* cct, std::string& path, uint64_t max_data_size);
 196   ~OpsLogFile() override;
 197   CephContext *get_cct() const override { return cct; }
 198   unsigned get_subsys() const override { return dout_subsys; }
 199   std::ostream& gen_prefix(std::ostream& out) const override { return out << "rgw OpsLogFile: "; }
 200   void reopen();
 201   void start();
 202   void stop();
 203 };
 204
 205 class OpsLogSocket : public OutputDataSocket, public JsonOpsLogSink {
 206 protected:
 207   int log_json(struct req_state* s, bufferlist& bl) override;
 208   void init_connection(bufferlist& bl) override;
 209
 210 public:
 211   OpsLogSocket(CephContext *cct, uint64_t _backlog);
 212 };
 213
 214 class OpsLogRados : public OpsLogSink {
 215   // main()'s Store pointer as a reference, possibly modified by RGWRealmReloader
 216   rgw::sal::Store* const& store;
 217
 218 public:
 219   OpsLogRados(rgw::sal::Store* const& store);
 220   int log(struct req_state* s, struct rgw_log_entry& entry) override;
 221 };
 222
 223 class RGWREST;
 224
 225 int rgw_log_op(RGWREST* const rest, struct req_state* s,
 226                const std::string& op_name, OpsLogSink* olog);
 227 void rgw_log_usage_init(CephContext* cct, rgw::sal::Store* store);
 228 void rgw_log_usage_finalize();
 229 void rgw_format_ops_log_entry(struct rgw_log_entry& entry,
 230                               ceph::Formatter *formatter);
 231
 232 #endif /* CEPH_RGW_LOG_H */
 233