diff options
Diffstat (limited to 'src/rgw/rgw_log.cc')
| -rw-r--r-- | src/rgw/rgw_log.cc | 683 |
1 files changed, 683 insertions, 0 deletions
diff --git a/src/rgw/rgw_log.cc b/src/rgw/rgw_log.cc new file mode 100644 index 000000000..f0509586c --- /dev/null +++ b/src/rgw/rgw_log.cc @@ -0,0 +1,683 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab ft=cpp + +#include "common/Clock.h" +#include "common/Timer.h" +#include "common/utf8.h" +#include "common/OutputDataSocket.h" +#include "common/Formatter.h" + +#include "rgw_bucket.h" +#include "rgw_log.h" +#include "rgw_acl.h" +#include "rgw_client_io.h" +#include "rgw_rest.h" +#include "rgw_zone.h" +#include "rgw_sal_rados.h" + +#include "services/svc_zone.h" + +#include <chrono> +#include <math.h> + +#define dout_subsys ceph_subsys_rgw + +static void set_param_str(struct req_state *s, const char *name, string& str) +{ + const char *p = s->info.env->get(name); + if (p) + str = p; +} + +string render_log_object_name(const string& format, + struct tm *dt, const string& bucket_id, + const string& bucket_name) +{ + string o; + for (unsigned i=0; i<format.size(); i++) { + if (format[i] == '%' && i+1 < format.size()) { + i++; + char buf[32]; + switch (format[i]) { + case '%': + strcpy(buf, "%"); + break; + case 'Y': + sprintf(buf, "%.4d", dt->tm_year + 1900); + break; + case 'y': + sprintf(buf, "%.2d", dt->tm_year % 100); + break; + case 'm': + sprintf(buf, "%.2d", dt->tm_mon + 1); + break; + case 'd': + sprintf(buf, "%.2d", dt->tm_mday); + break; + case 'H': + sprintf(buf, "%.2d", dt->tm_hour); + break; + case 'I': + sprintf(buf, "%.2d", (dt->tm_hour % 12) + 1); + break; + case 'k': + sprintf(buf, "%d", dt->tm_hour); + break; + case 'l': + sprintf(buf, "%d", (dt->tm_hour % 12) + 1); + break; + case 'M': + sprintf(buf, "%.2d", dt->tm_min); + break; + + case 'i': + o += bucket_id; + continue; + case 'n': + o += bucket_name; + continue; + default: + // unknown code + sprintf(buf, "%%%c", format[i]); + break; + } + o += buf; + continue; + } + o += format[i]; + } + return o; +} + +/* usage logger */ +class UsageLogger : public DoutPrefixProvider { + CephContext *cct; + RGWRados *store; + map<rgw_user_bucket, RGWUsageBatch> usage_map; + ceph::mutex lock = ceph::make_mutex("UsageLogger"); + int32_t num_entries; + ceph::mutex timer_lock = ceph::make_mutex("UsageLogger::timer_lock"); + SafeTimer timer; + utime_t round_timestamp; + + class C_UsageLogTimeout : public Context { + UsageLogger *logger; + public: + explicit C_UsageLogTimeout(UsageLogger *_l) : logger(_l) {} + void finish(int r) override { + logger->flush(); + logger->set_timer(); + } + }; + + void set_timer() { + timer.add_event_after(cct->_conf->rgw_usage_log_tick_interval, new C_UsageLogTimeout(this)); + } +public: + + UsageLogger(CephContext *_cct, RGWRados *_store) : cct(_cct), store(_store), num_entries(0), timer(cct, timer_lock) { + timer.init(); + std::lock_guard l{timer_lock}; + set_timer(); + utime_t ts = ceph_clock_now(); + recalc_round_timestamp(ts); + } + + ~UsageLogger() { + std::lock_guard l{timer_lock}; + flush(); + timer.cancel_all_events(); + timer.shutdown(); + } + + void recalc_round_timestamp(utime_t& ts) { + round_timestamp = ts.round_to_hour(); + } + + void insert_user(utime_t& timestamp, const rgw_user& user, rgw_usage_log_entry& entry) { + lock.lock(); + if (timestamp.sec() > round_timestamp + 3600) + recalc_round_timestamp(timestamp); + entry.epoch = round_timestamp.sec(); + bool account; + string u = user.to_str(); + rgw_user_bucket ub(u, entry.bucket); + real_time rt = round_timestamp.to_real_time(); + usage_map[ub].insert(rt, entry, &account); + if (account) + num_entries++; + bool need_flush = (num_entries > cct->_conf->rgw_usage_log_flush_threshold); + lock.unlock(); + if (need_flush) { + std::lock_guard l{timer_lock}; + flush(); + } + } + + void insert(utime_t& timestamp, rgw_usage_log_entry& entry) { + if (entry.payer.empty()) { + insert_user(timestamp, entry.owner, entry); + } else { + insert_user(timestamp, entry.payer, entry); + } + } + + void flush() { + map<rgw_user_bucket, RGWUsageBatch> old_map; + lock.lock(); + old_map.swap(usage_map); + num_entries = 0; + lock.unlock(); + + store->log_usage(this, old_map); + } + + CephContext *get_cct() const override { return cct; } + unsigned get_subsys() const override { return dout_subsys; } + std::ostream& gen_prefix(std::ostream& out) const override { return out << "rgw UsageLogger: "; } +}; + +static UsageLogger *usage_logger = NULL; + +void rgw_log_usage_init(CephContext *cct, RGWRados *store) +{ + usage_logger = new UsageLogger(cct, store); +} + +void rgw_log_usage_finalize() +{ + delete usage_logger; + usage_logger = NULL; +} + +static void log_usage(struct req_state *s, const string& op_name) +{ + if (s->system_request) /* don't log system user operations */ + return; + + if (!usage_logger) + return; + + rgw_user user; + rgw_user payer; + string bucket_name; + + bucket_name = s->bucket_name; + + if (!bucket_name.empty()) { + bucket_name = s->bucket_name; + user = s->bucket_owner.get_id(); + if (!rgw::sal::RGWBucket::empty(s->bucket.get()) && + s->bucket->get_info().requester_pays) { + payer = s->user->get_id(); + } + } else { + user = s->user->get_id(); + } + + bool error = s->err.is_err(); + if (error && s->err.http_ret == 404) { + bucket_name = "-"; /* bucket not found, use the invalid '-' as bucket name */ + } + + string u = user.to_str(); + string p = payer.to_str(); + rgw_usage_log_entry entry(u, p, bucket_name); + + uint64_t bytes_sent = ACCOUNTING_IO(s)->get_bytes_sent(); + uint64_t bytes_received = ACCOUNTING_IO(s)->get_bytes_received(); + + rgw_usage_data data(bytes_sent, bytes_received); + + data.ops = 1; + if (!s->is_err()) + data.successful_ops = 1; + + ldpp_dout(s, 30) << "log_usage: bucket_name=" << bucket_name + << " tenant=" << s->bucket_tenant + << ", bytes_sent=" << bytes_sent << ", bytes_received=" + << bytes_received << ", success=" << data.successful_ops << dendl; + + entry.add(op_name, data); + + utime_t ts = ceph_clock_now(); + + usage_logger->insert(ts, entry); +} + +void rgw_format_ops_log_entry(struct rgw_log_entry& entry, Formatter *formatter) +{ + formatter->open_object_section("log_entry"); + formatter->dump_string("bucket", entry.bucket); + { + auto t = utime_t{entry.time}; + t.gmtime(formatter->dump_stream("time")); // UTC + t.localtime(formatter->dump_stream("time_local")); + } + formatter->dump_string("remote_addr", entry.remote_addr); + string obj_owner = entry.object_owner.to_str(); + if (obj_owner.length()) + formatter->dump_string("object_owner", obj_owner); + formatter->dump_string("user", entry.user); + formatter->dump_string("operation", entry.op); + formatter->dump_string("uri", entry.uri); + formatter->dump_string("http_status", entry.http_status); + formatter->dump_string("error_code", entry.error_code); + formatter->dump_int("bytes_sent", entry.bytes_sent); + formatter->dump_int("bytes_received", entry.bytes_received); + formatter->dump_int("object_size", entry.obj_size); + { + using namespace std::chrono; + uint64_t total_time = duration_cast<milliseconds>(entry.total_time).count(); + formatter->dump_int("total_time", total_time); + } + formatter->dump_string("user_agent", entry.user_agent); + formatter->dump_string("referrer", entry.referrer); + if (entry.x_headers.size() > 0) { + formatter->open_array_section("http_x_headers"); + for (const auto& iter: entry.x_headers) { + formatter->open_object_section(iter.first.c_str()); + formatter->dump_string(iter.first.c_str(), iter.second); + formatter->close_section(); + } + formatter->close_section(); + } + formatter->dump_string("trans_id", entry.trans_id); + switch(entry.identity_type) { + case TYPE_RGW: + formatter->dump_string("authentication_type","Local"); + break; + case TYPE_LDAP: + formatter->dump_string("authentication_type","LDAP"); + break; + case TYPE_KEYSTONE: + formatter->dump_string("authentication_type","Keystone"); + break; + case TYPE_WEB: + formatter->dump_string("authentication_type","OIDC Provider"); + break; + case TYPE_ROLE: + formatter->dump_string("authentication_type","STS"); + break; + default: + break; + } + if (entry.token_claims.size() > 0) { + if (entry.token_claims[0] == "sts") { + formatter->open_object_section("sts_info"); + for (const auto& iter: entry.token_claims) { + auto pos = iter.find(":"); + if (pos != string::npos) { + formatter->dump_string(iter.substr(0, pos), iter.substr(pos + 1)); + } + } + formatter->close_section(); + } + } + if (!entry.access_key_id.empty()) { + formatter->dump_string("access_key_id", entry.access_key_id); + } + if (!entry.subuser.empty()) { + formatter->dump_string("subuser", entry.subuser); + } + formatter->dump_bool("temp_url", entry.temp_url); + + if (entry.op == "multi_object_delete") { + formatter->open_object_section("op_data"); + formatter->dump_int("num_ok", entry.delete_multi_obj_meta.num_ok); + formatter->dump_int("num_err", entry.delete_multi_obj_meta.num_err); + formatter->open_array_section("objects"); + for (const auto& iter: entry.delete_multi_obj_meta.objects) { + formatter->open_object_section(""); + formatter->dump_string("key", iter.key); + formatter->dump_string("version_id", iter.version_id); + formatter->dump_int("http_status", iter.http_status); + formatter->dump_bool("error", iter.error); + if (iter.error) { + formatter->dump_string("error_message", iter.error_message); + } else { + formatter->dump_bool("delete_marker", iter.delete_marker); + formatter->dump_string("marker_version_id", iter.marker_version_id); + } + formatter->close_section(); + } + formatter->close_section(); + formatter->close_section(); + } + formatter->close_section(); +} + +OpsLogManifold::~OpsLogManifold() +{ + for (const auto &sink : sinks) { + delete sink; + } +} + +void OpsLogManifold::add_sink(OpsLogSink* sink) +{ + sinks.push_back(sink); +} + +int OpsLogManifold::log(struct req_state* s, struct rgw_log_entry& entry) +{ + int ret = 0; + for (const auto &sink : sinks) { + if (sink->log(s, entry) < 0) { + ret = -1; + } + } + return ret; +} + +OpsLogFile::OpsLogFile(CephContext* cct, std::string& path, uint64_t max_data_size) : + cct(cct), data_size(0), max_data_size(max_data_size), path(path), need_reopen(false) +{ +} + +void OpsLogFile::reopen() { + need_reopen = true; +} + +void OpsLogFile::flush() +{ + { + std::scoped_lock log_lock(mutex); + assert(flush_buffer.empty()); + flush_buffer.swap(log_buffer); + data_size = 0; + } + for (auto bl : flush_buffer) { + int try_num = 0; + while (true) { + if (!file.is_open() || need_reopen) { + need_reopen = false; + file.close(); + file.open(path, std::ofstream::app); + } + bl.write_stream(file); + if (!file) { + ldpp_dout(this, 0) << "ERROR: failed to log RGW ops log file entry" << dendl; + file.clear(); + if (stopped) { + break; + } + int sleep_time_secs = std::min((int) pow(2, try_num), 60); + std::this_thread::sleep_for(std::chrono::seconds(sleep_time_secs)); + try_num++; + } else { + break; + } + } + } + flush_buffer.clear(); + file << std::endl; +} + +void* OpsLogFile::entry() { + std::unique_lock lock(mutex); + while (!stopped) { + if (!log_buffer.empty()) { + lock.unlock(); + flush(); + lock.lock(); + continue; + } + cond.wait(lock); + } + lock.unlock(); + flush(); + return NULL; +} + +void OpsLogFile::start() { + stopped = false; + create("ops_log_file"); +} + +void OpsLogFile::stop() { + { + std::unique_lock lock(mutex); + cond.notify_one(); + stopped = true; + } + join(); +} + +OpsLogFile::~OpsLogFile() +{ + if (!stopped) { + stop(); + } + file.close(); +} + +int OpsLogFile::log_json(struct req_state* s, bufferlist& bl) +{ + std::unique_lock lock(mutex); + if (data_size + bl.length() >= max_data_size) { + ldout(s->cct, 0) << "ERROR: RGW ops log file buffer too full, dropping log for txn: " << s->trans_id << dendl; + return -1; + } + log_buffer.push_back(bl); + data_size += bl.length(); + cond.notify_all(); + return 0; +} + +JsonOpsLogSink::JsonOpsLogSink() { + formatter = new JSONFormatter; +} + +JsonOpsLogSink::~JsonOpsLogSink() { + delete formatter; +} + +void JsonOpsLogSink::formatter_to_bl(bufferlist& bl) +{ + stringstream ss; + formatter->flush(ss); + const string& s = ss.str(); + bl.append(s); +} + +int JsonOpsLogSink::log(struct req_state* s, struct rgw_log_entry& entry) +{ + bufferlist bl; + + lock.lock(); + rgw_format_ops_log_entry(entry, formatter); + formatter_to_bl(bl); + lock.unlock(); + + return log_json(s, bl); +} + +void OpsLogSocket::init_connection(bufferlist& bl) +{ + bl.append("["); +} + +OpsLogSocket::OpsLogSocket(CephContext *cct, uint64_t _backlog) : OutputDataSocket(cct, _backlog) +{ + delim.append(",\n"); +} + +int OpsLogSocket::log_json(struct req_state* s, bufferlist& bl) +{ + append_output(bl); + return 0; +} + +OpsLogRados::OpsLogRados(rgw::sal::RGWRadosStore* const& store) : store(store) +{ +} + +int OpsLogRados::log(struct req_state* s, struct rgw_log_entry& entry) +{ + if (!s->cct->_conf->rgw_ops_log_rados) { + return 0; + } + bufferlist bl; + encode(entry, bl); + + struct tm bdt; + time_t t = req_state::Clock::to_time_t(entry.time); + if (s->cct->_conf->rgw_log_object_name_utc) + gmtime_r(&t, &bdt); + else + localtime_r(&t, &bdt); + + RGWRados* rados = store->getRados(); + + string oid = render_log_object_name(s->cct->_conf->rgw_log_object_name, &bdt, + entry.bucket_id, entry.bucket); + rgw_raw_obj obj(rados->svc.zone->get_zone_params().log_pool, oid); + int ret = rados->append_async(s, obj, bl.length(), bl); + if (ret == -ENOENT) { + ret = rados->create_pool(s, rados->svc.zone->get_zone_params().log_pool); + if (ret < 0) + goto done; + // retry + ret = rados->append_async(s, obj, bl.length(), bl); + } +done: + if (ret < 0) { + ldpp_dout(s, 0) << "ERROR: failed to log RADOS RGW ops log entry for txn: " << s->trans_id << dendl; + } + return ret; +} + +int rgw_log_op(RGWREST* const rest, req_state *s, const RGWOp* op, OpsLogSink *olog) +{ + struct rgw_log_entry entry; + string bucket_id; + string op_name = (op ? op->name() : "unknown"); + + if (s->enable_usage_log) + log_usage(s, op_name); + + if (!s->enable_ops_log) + return 0; + + if (s->bucket_name.empty()) { + ldpp_dout(s, 5) << "nothing to log for operation" << dendl; + return -EINVAL; + } + if (s->err.ret == -ERR_NO_SUCH_BUCKET || rgw::sal::RGWBucket::empty(s->bucket.get())) { + if (!s->cct->_conf->rgw_log_nonexistent_bucket) { + ldpp_dout(s, 5) << "bucket " << s->bucket_name << " doesn't exist, not logging" << dendl; + return 0; + } + bucket_id = ""; + } else { + bucket_id = s->bucket->get_bucket_id(); + } + entry.bucket = rgw_make_bucket_entry_name(s->bucket_tenant, s->bucket_name); + + if (check_utf8(entry.bucket.c_str(), entry.bucket.size()) != 0) { + ldpp_dout(s, 5) << "not logging op on bucket with non-utf8 name" << dendl; + return 0; + } + + if (!rgw::sal::RGWObject::empty(s->object.get())) { + entry.obj = s->object->get_key(); + } else { + entry.obj = rgw_obj_key("-"); + } + + entry.obj_size = s->obj_size; + + if (s->cct->_conf->rgw_remote_addr_param.length()) + set_param_str(s, s->cct->_conf->rgw_remote_addr_param.c_str(), + entry.remote_addr); + else + set_param_str(s, "REMOTE_ADDR", entry.remote_addr); + set_param_str(s, "HTTP_USER_AGENT", entry.user_agent); + // legacy apps are still using misspelling referer, such as curl -e option + if (s->info.env->exists("HTTP_REFERRER")) + set_param_str(s, "HTTP_REFERRER", entry.referrer); + else + set_param_str(s, "HTTP_REFERER", entry.referrer); + + std::string uri; + if (s->info.env->exists("REQUEST_METHOD")) { + uri.append(s->info.env->get("REQUEST_METHOD")); + uri.append(" "); + } + + if (s->info.env->exists("REQUEST_URI")) { + uri.append(s->info.env->get("REQUEST_URI")); + } + + if (s->info.env->exists("QUERY_STRING")) { + const char* qs = s->info.env->get("QUERY_STRING"); + if(qs && (*qs != '\0')) { + uri.append("?"); + uri.append(qs); + } + } + + if (s->info.env->exists("HTTP_VERSION")) { + uri.append(" "); + uri.append("HTTP/"); + uri.append(s->info.env->get("HTTP_VERSION")); + } + + entry.uri = std::move(uri); + + entry.op = op_name; + if (op) { + op->write_ops_log_entry(entry); + } + + if (s->auth.identity) { + entry.identity_type = s->auth.identity->get_identity_type(); + s->auth.identity->write_ops_log_entry(entry); + } else { + entry.identity_type = TYPE_NONE; + } + + if (! s->token_claims.empty()) { + entry.token_claims = std::move(s->token_claims); + } + + /* custom header logging */ + if (rest) { + if (rest->log_x_headers()) { + for (const auto& iter : s->info.env->get_map()) { + if (rest->log_x_header(iter.first)) { + entry.x_headers.insert( + rgw_log_entry::headers_map::value_type(iter.first, iter.second)); + } + } + } + } + + entry.user = s->user->get_id().to_str(); + if (s->object_acl) + entry.object_owner = s->object_acl->get_owner().get_id(); + entry.bucket_owner = s->bucket_owner.get_id(); + + uint64_t bytes_sent = ACCOUNTING_IO(s)->get_bytes_sent(); + uint64_t bytes_received = ACCOUNTING_IO(s)->get_bytes_received(); + + entry.time = s->time; + entry.total_time = s->time_elapsed(); + entry.bytes_sent = bytes_sent; + entry.bytes_received = bytes_received; + if (s->err.http_ret) { + char buf[16]; + snprintf(buf, sizeof(buf), "%d", s->err.http_ret); + entry.http_status = buf; + } else { + entry.http_status = "200"; // default + } + entry.error_code = s->err.err_code; + entry.bucket_id = bucket_id; + entry.trans_id = s->trans_id; + if (olog) { + return olog->log(s, entry); + } + return 0; +} |