From: ShreeJejurikar Date: Wed, 3 Jun 2026 11:29:33 +0000 (+0530) Subject: rgw: emit bucket logging records for lifecycle delete actions X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=efeb0fe52f921abaa6d60cb273674ec42819665d;p=ceph.git rgw: emit bucket logging records for lifecycle delete actions LC delete actions now emit LIFECYCLE.DELETE.OBJECT journal records via no-req_state overloads of the bucketlogging frontend. Signed-off-by: ShreeJejurikar --- diff --git a/src/rgw/radosgw-admin/radosgw-admin.cc b/src/rgw/radosgw-admin/radosgw-admin.cc index 2f2cbe9dc78..183b202be3a 100644 --- a/src/rgw/radosgw-admin/radosgw-admin.cc +++ b/src/rgw/radosgw-admin/radosgw-admin.cc @@ -7966,7 +7966,7 @@ int main(int argc, const char **argv) } std::string old_obj; const auto region = driver->get_zone()->get_zonegroup().get_api_name(); - ret = rgw::bucketlogging::rollover_logging_object(configuration, target_bucket, obj_name, dpp(), region, bucket, null_yield, true, &objv_tracker, false, &old_obj); + ret = rgw::bucketlogging::rollover_logging_object(configuration, target_bucket, obj_name, dpp(), region, bucket.get(), null_yield, true, &objv_tracker, false, &old_obj); if (ret < 0) { cerr << "ERROR: failed to flush pending logging object '" << obj_name << "' to target bucket '" << configuration.target_bucket << "'. error: " << cpp_strerror(-ret) << std::endl; diff --git a/src/rgw/rgw_bucket_logging.cc b/src/rgw/rgw_bucket_logging.cc index c7f46605876..3ca820ee214 100644 --- a/src/rgw/rgw_bucket_logging.cc +++ b/src/rgw/rgw_bucket_logging.cc @@ -5,6 +5,7 @@ #include #include "common/ceph_time.h" #include "rgw_bucket_logging.h" +#include "rgw_bucket_logging_types.h" #include "rgw_xml.h" #include "rgw_sal.h" #include "rgw_op.h" @@ -239,7 +240,7 @@ ceph::coarse_real_time time_from_name(const std::string& obj_name, const DoutPre return extracted_time; } -std::string full_bucket_name(const std::unique_ptr& bucket) { +std::string full_bucket_name(const rgw::sal::Bucket* bucket) { if (bucket->get_tenant().empty()) { return bucket->get_name(); } @@ -251,7 +252,7 @@ int new_logging_object(const configuration& conf, std::string& obj_name, const DoutPrefixProvider *dpp, const std::string& region, - const std::unique_ptr& source_bucket, + rgw::sal::Bucket* source_bucket, optional_yield y, std::optional old_name, RGWObjVersionTracker* objv_tracker) { @@ -347,7 +348,7 @@ int rollover_logging_object(const configuration& conf, std::string& obj_name, const DoutPrefixProvider *dpp, const std::string& region, - const std::unique_ptr& source_bucket, + rgw::sal::Bucket* source_bucket, optional_yield y, bool must_commit, RGWObjVersionTracker* objv_tracker, @@ -465,16 +466,73 @@ int log_record(rgw::sal::Driver* driver, optional_yield y, bool async_completion, bool log_source_bucket) { - if (!s->bucket) { - ldpp_dout(dpp, 1) << "ERROR: only bucket operations are logged in bucket logging" << dendl; - return -EINVAL; // this should never happen + record_input input; + input.bucket = s->bucket.get(); + input.time = s->time; + if (obj) { + input.version_id = obj->get_instance(); + } + input.journal_err_out = &s->err.message; + // AWS S3 spec: log the assumed-role ARN for STS-credentialed requests. + if (s->auth.identity && s->auth.identity->get_identity_type() == TYPE_ROLE) { + if (auto caller_arn = s->auth.identity->get_caller_identity(); caller_arn) { + input.user_or_account = caller_arn->to_string(); + } + } + if (input.user_or_account.empty()) { + if (s->account_name.empty()) { + s->user->get_id().to_str(input.user_or_account); + } else { + input.user_or_account = s->account_name; + } } - auto set_journal_err = [&conf, s](const std::string& err_message) { - if (conf.logging_type == LoggingType::Journal) s->err.message = err_message; + input.fqdn = s->info.host; + if (!s->info.domain.empty() && !input.fqdn.empty()) { + input.fqdn.append(".").append(s->info.domain); + } + rgw::auth::s3::get_aws_version_and_auth_type(s, input.aws_version, input.auth_type); + input.time_elapsed = std::chrono::duration_cast(s->time_elapsed()); + input.remote_addr = s->info.env->get("REMOTE_ADDR", "-"); + input.trans_id = s->trans_id; + input.method = s->info.method; + input.request_uri = s->info.request_uri; + input.request_params = s->info.request_params; + input.referrer = s->info.env->get("HTTP_REFERER", "-"); + input.user_agent = s->info.env->get("HTTP_USER_AGENT", "-"); + input.ssl_cipher = s->info.env->get("SSL_CIPHER", "-"); + input.tls_version = s->info.env->get("TLS_VERSION", "-"); + input.x_amz_id_2 = s->info.x_meta_map.contains("x-amz-id-2") ? + s->info.x_meta_map.at("x-amz-id-2") : "-"; + input.http_ret = s->err.http_ret; + input.err_code = s->err.err_code; + input.content_length = s->content_length; + input.granted_by_acl = s->granted_by_acl; + input.src_object = s->src_object.get(); + input.src_bucket_name = s->src_bucket_name; + return log_record(driver, obj, input, op_name, etag, size, conf, + dpp, y, async_completion, log_source_bucket); +} + +int log_record(rgw::sal::Driver* driver, + const sal::Object* obj, + const record_input& input, + const std::string& op_name, + const std::string& etag, + size_t size, + const configuration& conf, + const DoutPrefixProvider *dpp, + optional_yield y, + bool async_completion, + bool log_source_bucket) { + auto set_journal_err = [&conf, &input](const std::string& err_message) { + if (conf.logging_type == LoggingType::Journal && input.journal_err_out) { + *input.journal_err_out = err_message; + } }; std::string target_bucket_name; std::string target_tenant_name; - int ret = rgw_parse_url_bucket(conf.target_bucket, s->bucket_tenant, target_tenant_name, target_bucket_name); + int ret = rgw_parse_url_bucket(conf.target_bucket, input.bucket->get_tenant(), + target_tenant_name, target_bucket_name); if (ret < 0) { ldpp_dout(dpp, 1) << "ERROR: failed to parse logging bucket name '" << conf.target_bucket << "', ret = " << ret << dendl; set_journal_err(fmt::format("Failed to parse logging bucket name '{}'", conf.target_bucket)); @@ -492,7 +550,11 @@ int log_record(rgw::sal::Driver* driver, rgw::ARN target_resource_arn(target_bucket_id, conf.target_prefix); std::string err_message; - if (ret = verify_target_bucket_policy(dpp, target_bucket.get(), target_resource_arn, s, &err_message); ret < 0) { + if (ret = verify_target_bucket_policy(dpp, target_bucket.get(), target_resource_arn, + dpp->get_cct(), + rgw::ARN(input.bucket->get_key()).to_string(), + to_string(input.bucket->get_owner()), + &err_message); ret < 0) { set_journal_err(err_message); return ret; } @@ -503,9 +565,9 @@ int log_record(rgw::sal::Driver* driver, } // make sure that the logging source attribute is up-to-date - if (ret = update_bucket_logging_sources(dpp, target_bucket, s->bucket->get_key(), true, y); ret < 0) { + if (ret = update_bucket_logging_sources(dpp, target_bucket, input.bucket->get_key(), true, y); ret < 0) { ldpp_dout(dpp, 5) << "WARNING: could not update bucket logging source '" << - s->bucket->get_key() << "' in logging bucket '" << target_bucket_id << "' attribute, during record logging. ret = " << ret << dendl; + input.bucket->get_key() << "' in logging bucket '" << target_bucket_id << "' attribute, during record logging. ret = " << ret << dendl; } const auto region = driver->get_zone()->get_zonegroup().get_api_name(); @@ -517,7 +579,7 @@ int log_record(rgw::sal::Driver* driver, if (ceph::coarse_real_time::clock::now() > time_to_commit) { ldpp_dout(dpp, 20) << "INFO: logging object '" << obj_name << "' exceeded its time, will be committed to logging bucket '" << target_bucket_id << "'" << dendl; - if (ret = rollover_logging_object(conf, target_bucket, obj_name, dpp, region, s->bucket, y, false, &objv_tracker, true, nullptr, &err_message); ret < 0 && ret != -ECANCELED) { + if (ret = rollover_logging_object(conf, target_bucket, obj_name, dpp, region, input.bucket, y, false, &objv_tracker, true, nullptr, &err_message); ret < 0 && ret != -ECANCELED) { set_journal_err(err_message); return ret; } @@ -526,7 +588,7 @@ int log_record(rgw::sal::Driver* driver, } } else if (ret == -ENOENT) { // try to create the temporary log object for the first time - ret = new_logging_object(conf, target_bucket, obj_name, dpp, region, s->bucket, y, std::nullopt, &objv_tracker); + ret = new_logging_object(conf, target_bucket, obj_name, dpp, region, input.bucket, y, std::nullopt, &objv_tracker); if (ret == 0) { ldpp_dout(dpp, 20) << "INFO: first time logging for logging bucket '" << target_bucket_id << "' and prefix '" << conf.target_prefix << "'" << dendl; @@ -547,44 +609,23 @@ int log_record(rgw::sal::Driver* driver, } std::string record; - const auto tt = ceph::coarse_real_time::clock::to_time_t(s->time); + const auto tt = ceph::coarse_real_time::clock::to_time_t(input.time); std::tm t{}; localtime_r(&tt, &t); - // AWS S3 spec: log the assumed-role ARN for STS-credentialed requests. - std::string user_or_account; - if (s->auth.identity && s->auth.identity->get_identity_type() == TYPE_ROLE) { - if (auto caller_arn = s->auth.identity->get_caller_identity(); caller_arn) { - user_or_account = caller_arn->to_string(); - } - } - if (user_or_account.empty()) { - if (s->account_name.empty()) { - s->user->get_id().to_str(user_or_account); - } else { - user_or_account = s->account_name; - } - } - auto fqdn = s->info.host; - if (!s->info.domain.empty() && !fqdn.empty()) { - fqdn.append(".").append(s->info.domain); - } - std::string aws_version("-"); - std::string auth_type("-"); - rgw::auth::s3::get_aws_version_and_auth_type(s, aws_version, auth_type); std::string bucket_owner; std::string bucket_name; if (log_source_bucket && conf.logging_type == LoggingType::Standard) { // log source bucket for COPY operations only in standard mode - if (!s->src_object || !s->src_object->get_bucket()) { + if (!input.src_object || !input.src_object->get_bucket()) { ldpp_dout(dpp, 1) << "ERROR: source object or bucket is missing when logging source bucket" << dendl; return -EINVAL; } - bucket_owner = to_string(s->src_object->get_bucket()->get_owner()); - bucket_name = s->src_bucket_name; + bucket_owner = to_string(input.src_object->get_bucket()->get_owner()); + bucket_name = input.src_bucket_name; } else { - bucket_owner = to_string(s->bucket->get_owner()); - bucket_name = full_bucket_name(s->bucket); + bucket_owner = to_string(input.bucket->get_owner()); + bucket_name = full_bucket_name(input.bucket); } switch (conf.logging_type) { @@ -593,32 +634,32 @@ int log_record(rgw::sal::Driver* driver, dash_if_empty(bucket_owner), dash_if_empty(bucket_name), t, - s->info.env->get("REMOTE_ADDR", "-"), - dash_if_empty(user_or_account), - dash_if_empty(s->trans_id), + dash_if_empty(input.remote_addr), + dash_if_empty(input.user_or_account), + dash_if_empty(input.trans_id), op_name, dash_if_empty_or_null(obj, obj->get_name()), - s->info.method, - s->info.request_uri, - s->info.request_params.empty() ? "" : "?", - s->info.request_params, - dash_if_zero(s->err.http_ret), - dash_if_empty(s->err.err_code), - dash_if_zero(s->content_length), + input.method, + input.request_uri, + input.request_params.empty() ? "" : "?", + input.request_params, + dash_if_zero(input.http_ret), + dash_if_empty(input.err_code), + dash_if_zero(input.content_length), dash_if_zero(size), "-", // no total time when logging record - std::chrono::duration_cast(s->time_elapsed()), - s->info.env->get("HTTP_REFERER", "-"), - s->info.env->get("HTTP_USER_AGENT", "-"), - dash_if_empty_or_null(obj, obj->get_instance()), - s->info.x_meta_map.contains("x-amz-id-2") ? s->info.x_meta_map.at("x-amz-id-2") : "-", - aws_version, - s->info.env->get("SSL_CIPHER", "-"), - dash_if_empty(auth_type), - dash_if_empty(fqdn), - s->info.env->get("TLS_VERSION", "-"), + input.time_elapsed, + dash_if_empty(input.referrer), + dash_if_empty(input.user_agent), + dash_if_empty(input.version_id), + dash_if_empty(input.x_amz_id_2), + dash_if_empty(input.aws_version), + dash_if_empty(input.ssl_cipher), + dash_if_empty(input.auth_type), + dash_if_empty(input.fqdn), + dash_if_empty(input.tls_version), "-", // no access point ARN - (s->granted_by_acl) ? "Yes" : "-"); + (input.granted_by_acl) ? "Yes" : "-"); break; case LoggingType::Journal: record = fmt::format("{} {} [{:%d/%b/%Y:%H:%M:%S %z}] {} {} {} {} {}", @@ -628,7 +669,7 @@ int log_record(rgw::sal::Driver* driver, op_name, dash_if_empty_or_null(obj, obj->get_name()), dash_if_zero(size), - dash_if_empty_or_null(obj, obj->get_instance()), + dash_if_empty(input.version_id), dash_if_empty(etag)); break; case LoggingType::Any: @@ -679,7 +720,7 @@ int log_record(rgw::sal::Driver* driver, if (ret == -EFBIG) { ldpp_dout(dpp, 5) << "WARNING: logging object '" << obj_name << "' is full, will be committed to logging bucket '" << target_bucket->get_key() << "'" << dendl; - if (ret = rollover_logging_object(conf, target_bucket, obj_name, dpp, region, s->bucket, y, true, &objv_tracker, true, nullptr, &err_message); ret < 0 && ret != -ECANCELED) { + if (ret = rollover_logging_object(conf, target_bucket, obj_name, dpp, region, input.bucket, y, true, &objv_tracker, true, nullptr, &err_message); ret < 0 && ret != -ECANCELED) { set_journal_err(err_message); return ret; } @@ -736,7 +777,9 @@ int log_record(rgw::sal::Driver* driver, return 0; } if (configuration.key_filter.has_content()) { - if (!match(configuration.key_filter, obj->get_name())) { + // No-op (rather than crash) for bucket-level ops with no object: can't + // tell if a missing key matches the filter. + if (!obj || !match(configuration.key_filter, obj->get_name())) { return 0; } } @@ -747,6 +790,57 @@ int log_record(rgw::sal::Driver* driver, "'. ret=" << ret << dendl; return ret; } + } catch (buffer::error& err) { + ldpp_dout(dpp, 1) << "ERROR: failed to decode logging attribute '" << RGW_ATTR_BUCKET_LOGGING + << "'. error: " << err.what() << dendl; + return -EINVAL; + } + return 0; +} + +int log_record(rgw::sal::Driver* driver, + LoggingType type, + const sal::Object* obj, + const record_input& input, + const std::string& op_name, + const std::string& etag, + size_t size, + const DoutPrefixProvider *dpp, + optional_yield y, + bool async_completion, + bool log_source_bucket) { + if (!input.bucket) { + ldpp_dout(dpp, 1) << "ERROR: only bucket operations are logged in bucket logging" << dendl; + return -EINVAL; + } + // check if bucket logging is needed + const auto& bucket_attrs = input.bucket->get_attrs(); + auto iter = bucket_attrs.find(RGW_ATTR_BUCKET_LOGGING); + if (iter == bucket_attrs.end()) { + return 0; + } + configuration configuration; + try { + configuration.enabled = true; + auto bl_iter = iter->second.cbegin(); + decode(configuration, bl_iter); + if (type != LoggingType::Any && configuration.logging_type != type) { + return 0; + } + if (configuration.key_filter.has_content()) { + // No-op (rather than crash) for bucket-level ops with no object: can't + // tell if a missing key matches the filter. + if (!obj || !match(configuration.key_filter, obj->get_name())) { + return 0; + } + } + ldpp_dout(dpp, 20) << "INFO: found matching logging configuration of bucket '" << input.bucket->get_key() << + "' configuration: " << configuration.to_json_str() << dendl; + if (const int ret = log_record(driver, obj, input, op_name, etag, size, configuration, dpp, y, async_completion, log_source_bucket); ret < 0) { + ldpp_dout(dpp, 1) << "ERROR: failed to perform logging for bucket '" << input.bucket->get_key() << + "'. ret=" << ret << dendl; + return ret; + } } catch (buffer::error& err) { ldpp_dout(dpp, 1) << "ERROR: failed to decode logging attribute '" << RGW_ATTR_BUCKET_LOGGING << "'. error: " << err.what() << dendl; @@ -977,7 +1071,9 @@ int source_bucket_cleanup(const DoutPrefixProvider* dpp, int verify_target_bucket_policy(const DoutPrefixProvider* dpp, rgw::sal::Bucket* target_bucket, const rgw::ARN& target_resource_arn, - req_state* s, + CephContext* cct, + std::string source_bucket_arn, + std::string source_account, std::string* err_message) { // verify target permissions for bucket logging // this is implementing the policy based permission granting from: @@ -993,19 +1089,20 @@ int verify_target_bucket_policy(const DoutPrefixProvider* dpp, return -EACCES; } try { - const rgw::IAM::Policy policy{s->cct, &target_bucket_id.tenant, policy_it->second.to_str(), false}; + const rgw::IAM::Policy policy{cct, &target_bucket_id.tenant, policy_it->second.to_str(), false}; ldpp_dout(dpp, 20) << "INFO: logging bucket '" << target_bucket_id << "' policy: " << policy << dendl; rgw::auth::ServiceIdentity ident(rgw::bucketlogging::service_principal); - const auto source_bucket_arn = rgw::ARN(s->bucket->get_key()).to_string(); - const auto source_account = to_string(s->bucket_owner.id); - s->env.emplace("aws:SourceArn", source_bucket_arn); - s->env.emplace("aws:SourceAccount", source_account); - if (policy.eval(s->env, ident, rgw::IAM::s3PutObject, target_resource_arn) != rgw::IAM::Effect::Allow) { + // The AWS-spec policy for the logging service principal references only + // aws:SourceArn and aws:SourceAccount. Build a minimal env to match. + rgw::IAM::Environment env; + const auto arn_it = env.emplace("aws:SourceArn", std::move(source_bucket_arn)); + const auto acct_it = env.emplace("aws:SourceAccount", std::move(source_account)); + if (policy.eval(env, ident, rgw::IAM::s3PutObject, target_resource_arn) != rgw::IAM::Effect::Allow) { ldpp_dout(dpp, 1) << "ERROR: logging bucket: '" << target_bucket_id << "' must have a bucket policy that allows logging service principal to put objects in the following resource ARN: '" << - target_resource_arn.to_string() << "' from source bucket ARN: '" << source_bucket_arn << - "' and source account: '" << source_account << "'" << dendl; + target_resource_arn.to_string() << "' from source bucket ARN: '" << arn_it->second << + "' and source account: '" << acct_it->second << "'" << dendl; if (err_message) { *err_message = fmt::format("Logging bucket '{}' policy does not allow logging", target_bucket->get_name()); } @@ -1022,6 +1119,18 @@ int verify_target_bucket_policy(const DoutPrefixProvider* dpp, return 0; } +int verify_target_bucket_policy(const DoutPrefixProvider* dpp, + rgw::sal::Bucket* target_bucket, + const rgw::ARN& target_resource_arn, + req_state* s, + std::string* err_message) { + return verify_target_bucket_policy(dpp, target_bucket, target_resource_arn, + s->cct, + rgw::ARN(s->bucket->get_key()).to_string(), + to_string(s->bucket->get_owner()), + err_message); +} + int verify_target_bucket_attributes(const DoutPrefixProvider* dpp, rgw::sal::Bucket* target_bucket, std::string* err_message) { const auto& target_info = target_bucket->get_info(); if (target_info.requester_pays) { diff --git a/src/rgw/rgw_bucket_logging.h b/src/rgw/rgw_bucket_logging.h index 6d5b70bc9c9..344d6ddcd9f 100644 --- a/src/rgw/rgw_bucket_logging.h +++ b/src/rgw/rgw_bucket_logging.h @@ -158,6 +158,11 @@ inline std::string to_string(const Records& records) { return str_records; } +// Per-record input metadata for log_record(). Full definition lives in +// rgw_bucket_logging_types.h; callers that only pass it by reference need +// just this forward declaration. +struct record_input; + // log a bucket logging record according to the configuration int log_record(rgw::sal::Driver* driver, const sal::Object* obj, @@ -171,6 +176,19 @@ int log_record(rgw::sal::Driver* driver, bool async_completion, bool log_source_bucket); +// no-req_state variant of the log_record(req_state*, ..., configuration&) overload above +int log_record(rgw::sal::Driver* driver, + const sal::Object* obj, + const record_input& input, + const std::string& op_name, + const std::string& etag, + size_t size, + const configuration& conf, + const DoutPrefixProvider *dpp, + optional_yield y, + bool async_completion, + bool log_source_bucket); + // commit the pending log objec to the log bucket // and create a new pending log object // if "must_commit" is "false" the function will return success even if the pending log object was not committed @@ -181,7 +199,7 @@ int rollover_logging_object(const configuration& conf, std::string& obj_name, const DoutPrefixProvider *dpp, const std::string& region, - const std::unique_ptr& source_bucket, + rgw::sal::Bucket* source_bucket, optional_yield y, bool must_commit, RGWObjVersionTracker* objv_tracker, @@ -209,6 +227,19 @@ int log_record(rgw::sal::Driver* driver, bool async_completion, bool log_source_bucket); +// no-req_state variant of the log_record(LoggingType, ..., req_state*) overload above +int log_record(rgw::sal::Driver* driver, + LoggingType type, + const sal::Object* obj, + const record_input& input, + const std::string& op_name, + const std::string& etag, + size_t size, + const DoutPrefixProvider *dpp, + optional_yield y, + bool async_completion, + bool log_source_bucket); + // return (by ref) an rgw_bucket object with the bucket name and tenant name // fails if the bucket name is not in the format: [tenant name:] int get_bucket_id(const std::string& bucket_name, const std::string& tenant_name, rgw_bucket& bucket_id); @@ -252,6 +283,17 @@ int verify_target_bucket_policy(const DoutPrefixProvider* dpp, req_state* s, std::string* err_message = nullptr); +// no-req_state variant of the verify_target_bucket_policy(req_state*) overload above. +// source_bucket_arn and source_account are the ARN of the source bucket +// and the id of its owner; they are used as aws:SourceArn / aws:SourceAccount when evaluating the target (log) bucket's policy +int verify_target_bucket_policy(const DoutPrefixProvider* dpp, + rgw::sal::Bucket* target_bucket, + const rgw::ARN& target_resource_arn, + CephContext* cct, + std::string source_bucket_arn, + std::string source_account, + std::string* err_message = nullptr); + // verify that target bucket does not have: // - bucket logging // - requester pays diff --git a/src/rgw/rgw_bucket_logging_types.h b/src/rgw/rgw_bucket_logging_types.h new file mode 100644 index 00000000000..46db8a8e7f7 --- /dev/null +++ b/src/rgw/rgw_bucket_logging_types.h @@ -0,0 +1,59 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*- +// vim: ts=8 sw=2 sts=2 expandtab ft=cpp + +#pragma once + +#include +#include +#include +#include +#include "common/ceph_time.h" +#include "rgw_sal_fwd.h" + +namespace rgw::bucketlogging { + +// Per-record input metadata for log_record(). The req_state* overloads populate +// this from the request state; no-req_state callers (e.g. the lifecycle thread) +// build it directly. Non-owning views point into request-lifetime storage. +struct record_input { + rgw::sal::Bucket* bucket{nullptr}; + // Requester field for Standard records. HTTP callers: assumed-role ARN, + // account name, or user id. LC callers: the source bucket's owner. + // std::string (not view) because the populate code builds it in place. + std::string user_or_account; + ceph::coarse_real_time time; + // Object's version id for the record. HTTP callers derive this from the + // request's object; LC callers pass the pre-deletion instance. + std::string_view version_id; + // Where to write the Journal-mode error message on failure. HTTP callers + // point at s->err.message; no-req_state callers leave it null. + std::string* journal_err_out{nullptr}; + + // Standard-mode-only fields. Views point at request-lifetime storage in + // req_state / env_map. The four std::string fields below are built in + // place by the populate code and therefore cannot be views. + std::chrono::milliseconds time_elapsed{}; + std::string_view remote_addr; + std::string_view trans_id; + std::string_view method; + std::string_view request_uri; + std::string_view request_params; + std::string_view referrer; + std::string_view user_agent; + std::string_view ssl_cipher; + std::string_view tls_version; + std::string fqdn; // appended in place with the domain suffix + std::string_view x_amz_id_2; + std::string aws_version; // written by get_aws_version_and_auth_type + std::string auth_type; // written by get_aws_version_and_auth_type + int http_ret{0}; + std::string_view err_code; + size_t content_length{0}; + bool granted_by_acl{false}; + + // For COPY operations (used only with log_source_bucket=true, Standard mode). + const sal::Object* src_object{nullptr}; + std::string_view src_bucket_name; +}; + +} // namespace rgw::bucketlogging diff --git a/src/rgw/rgw_lc.cc b/src/rgw/rgw_lc.cc index fdfbbc5abf2..8b6adba2b8c 100644 --- a/src/rgw/rgw_lc.cc +++ b/src/rgw/rgw_lc.cc @@ -13,6 +13,7 @@ #include #include #include +#include #include "include/scope_guard.h" #include "include/function2.hpp" @@ -34,6 +35,8 @@ #include "rgw_sal.h" #include "rgw_multi_del.h" #include "rgw_multipart_meta_filter.h" +#include "rgw_bucket_logging.h" +#include "rgw_bucket_logging_types.h" #include "fmt/format.h" @@ -704,6 +707,39 @@ static void send_notification(const DoutPrefixProvider* dpp, } } +// Op name used in journal records for LC-driven deletions. +static const std::string lifecycle_delete_object_op = "LIFECYCLE.DELETE.OBJECT"; + +// LC has no req_state, so the source bucket's owner stands in as the +// identity for the log-bucket write permission check. +static void send_log_record(const DoutPrefixProvider* dpp, + optional_yield y, + rgw::sal::Driver* driver, + rgw::sal::Object* obj, + rgw::sal::Bucket* bucket, + const std::string& etag, + uint64_t size, + const std::string& version_id, + const std::string& op_name) { + rgw::bucketlogging::record_input input; + input.bucket = bucket; + input.user_or_account = to_string(bucket->get_owner()); + input.time = ceph::coarse_real_time::clock::now(); + input.version_id = version_id; + + const int ret = rgw::bucketlogging::log_record( + driver, + rgw::bucketlogging::LoggingType::Journal, + obj, input, op_name, etag, size, dpp, y, + /*async_completion=*/true, + /*log_source_bucket=*/false); + if (ret < 0) { + ldpp_dout(dpp, 1) << "WARNING: bucket logging failed for lc object: " + << obj->get_name() << " op: " << op_name + << " ret: " << ret << dendl; + } +} + /* do all zones in the zone group process LC? */ static bool zonegroup_lc_check(const DoutPrefixProvider *dpp, rgw::sal::Zone* zone) { @@ -729,7 +765,8 @@ static bool zonegroup_lc_check(const DoutPrefixProvider *dpp, rgw::sal::Zone* zo static int remove_expired_obj(const DoutPrefixProvider* dpp, optional_yield y, lc_op_ctx& oc, bool remove_indeed, - const rgw::notify::EventTypeList& event_types) { + const rgw::notify::EventTypeList& event_types, + boost::optional log_op_name = boost::none) { int ret{0}; auto& driver = oc.driver; auto& bucket_info = oc.bucket->get_info(); @@ -764,7 +801,8 @@ static int remove_expired_obj(const DoutPrefixProvider* dpp, } auto have_notify = !event_types.empty(); - if (have_notify) { + // etag is needed for both notifications and bucket-logging records + if (have_notify || log_op_name) { auto attrset = obj->get_attrs(); auto iter = attrset.find(RGW_ATTR_ETAG); if (iter != attrset.end()) { @@ -796,6 +834,10 @@ static int remove_expired_obj(const DoutPrefixProvider* dpp, send_notification(dpp, y, driver, obj.get(), oc.bucket, etag, size, version_id, event_types); } + if (log_op_name) { + send_log_record(dpp, y, driver, obj.get(), oc.bucket, etag, size, + version_id, *log_op_name); + } } return ret; @@ -1300,7 +1342,8 @@ public: /* ! o.is_delete_marker() */ r = remove_expired_obj(oc.dpp, y, oc, !oc.bucket->versioning_enabled(), {rgw::notify::ObjectExpirationCurrent, - rgw::notify::LifecycleExpirationDelete}); + rgw::notify::LifecycleExpirationDelete}, + lifecycle_delete_object_op); if (r < 0) { ldpp_dout(oc.dpp, 0) << "ERROR: remove_expired_obj " << oc.bucket << ":" << o.key @@ -1357,7 +1400,8 @@ public: auto& o = oc.o; int r = remove_expired_obj(oc.dpp, y, oc, true, {rgw::notify::LifecycleExpirationDelete, - rgw::notify::ObjectExpirationNoncurrent}); + rgw::notify::ObjectExpirationNoncurrent}, + lifecycle_delete_object_op); if (r < 0) { ldpp_dout(oc.dpp, 0) << "ERROR: remove_expired_obj (non-current expiration) " << oc.bucket << ":" << o.key @@ -1488,20 +1532,23 @@ public: */ if (! oc.bucket->versioning_enabled()) { ret = - remove_expired_obj(oc.dpp, y, oc, true, {/* no delete notify expected */}); + remove_expired_obj(oc.dpp, y, oc, true, {/* no delete notify expected */}, + lifecycle_delete_object_op); ldpp_dout(oc.dpp, 20) << "delete_tier_obj Object(key:" << oc.o.key << ") not versioned flags: " << oc.o.flags << dendl; } else { /* versioned */ if (oc.o.is_current() && !oc.o.is_delete_marker()) { - ret = remove_expired_obj(oc.dpp, y, oc, false, {/* no delete notify expected */}); + ret = remove_expired_obj(oc.dpp, y, oc, false, {/* no delete notify expected */}, + lifecycle_delete_object_op); ldpp_dout(oc.dpp, 20) << "delete_tier_obj Object(key:" << oc.o.key << ") current & not delete_marker" << " versioned_epoch: " << oc.o.versioned_epoch << "flags: " << oc.o.flags << dendl; } else { ret = remove_expired_obj(oc.dpp, y, oc, true, - {/* no delete notify expected */}); + {/* no delete notify expected */}, + lifecycle_delete_object_op); ldpp_dout(oc.dpp, 20) << "delete_tier_obj Object(key:" << oc.o.key << ") not current " << "versioned_epoch: " << oc.o.versioned_epoch diff --git a/src/rgw/rgw_rest_bucket_logging.cc b/src/rgw/rgw_rest_bucket_logging.cc index 9aa03204c3c..60c74946b21 100644 --- a/src/rgw/rgw_rest_bucket_logging.cc +++ b/src/rgw/rgw_rest_bucket_logging.cc @@ -323,7 +323,7 @@ class RGWPutBucketLoggingOp : public RGWDefaultResponseOp { obj_name, this, region, - src_bucket, + src_bucket.get(), y, false, // rollover should happen even if commit failed &objv_tracker, @@ -436,7 +436,7 @@ class RGWPostBucketLoggingOp : public RGWDefaultResponseOp { ldpp_dout(this, 5) << "INFO: no pending logging object in logging bucket '" << target_bucket_id << "'. new object should be created" << dendl; } const auto region = driver->get_zone()->get_zonegroup().get_api_name(); - op_ret = rgw::bucketlogging::rollover_logging_object(configuration, target_bucket, obj_name, this, region, source_bucket, y, true, &objv_tracker, false, &old_obj); + op_ret = rgw::bucketlogging::rollover_logging_object(configuration, target_bucket, obj_name, this, region, source_bucket.get(), y, true, &objv_tracker, false, &old_obj); if (op_ret < 0) { ldpp_dout(this, 1) << "ERROR: failed to flush pending logging object '" << obj_name << "'" << " to logging bucket '" << target_bucket_id << "'. " diff --git a/src/test/rgw/bucket_logging/test_bucket_logging.py b/src/test/rgw/bucket_logging/test_bucket_logging.py index f85aec6bc89..9e87c30e86c 100644 --- a/src/test/rgw/bucket_logging/test_bucket_logging.py +++ b/src/test/rgw/bucket_logging/test_bucket_logging.py @@ -273,6 +273,120 @@ def cleanup_bucket(s3_client, bucket_name): log.warning(f"Error cleaning up bucket {bucket_name}: {e}") +def cleanup_versioned_bucket(s3_client, bucket_name): + """Delete all object versions and delete-markers, then the bucket.""" + try: + resp = s3_client.list_object_versions(Bucket=bucket_name) + for v in resp.get('Versions', []): + s3_client.delete_object(Bucket=bucket_name, Key=v['Key'], VersionId=v['VersionId']) + for d in resp.get('DeleteMarkers', []): + s3_client.delete_object(Bucket=bucket_name, Key=d['Key'], VersionId=d['VersionId']) + s3_client.delete_bucket(Bucket=bucket_name) + log.debug(f"Deleted versioned bucket: {bucket_name}") + except ClientError as e: + log.warning(f"Error cleaning up versioned bucket {bucket_name}: {e}") + + +def ceph(args, **kwargs): + cmd = [test_path + 'test-rgw-call.sh', 'call_ceph', 'noname'] + args + return bash(cmd, **kwargs) + + +def set_lc_debug_interval(seconds): + # Each "1 day" in an LC rule becomes `seconds` real seconds. + return ceph(['config', 'set', 'client', 'rgw_lc_debug_interval', str(seconds)]) + + +def trigger_lc_processing(): + return admin(['lc', 'process']) + + +def wait_for_object_gone(s3_client, bucket, key, timeout=30, interval=1): + deadline = time.time() + timeout + while time.time() < deadline: + resp = s3_client.list_objects_v2(Bucket=bucket, Prefix=key) + if not any(c['Key'] == key for c in resp.get('Contents', [])): + return True + time.sleep(interval) + return False + + +def enable_versioning(s3_client, bucket): + s3_client.put_bucket_versioning( + Bucket=bucket, + VersioningConfiguration={'Status': 'Enabled'}, + ) + + +def apply_lc_config(s3_client, bucket, rules): + s3_client.put_bucket_lifecycle_configuration( + Bucket=bucket, + LifecycleConfiguration={'Rules': rules}, + ) + + +def make_lc_rule(prefix='', rule_id='rule', **action): + return { + 'ID': rule_id, + 'Status': 'Enabled', + 'Filter': {'Prefix': prefix}, + **action, + } + + +def create_orphan_delete_marker(s3_client, bucket, key): + v1 = s3_client.put_object(Bucket=bucket, Key=key, Body=b'data')['VersionId'] + s3_client.delete_object(Bucket=bucket, Key=key) + s3_client.delete_object(Bucket=bucket, Key=key, VersionId=v1) + + +# {bucket_owner} {bucket_name} [{date}] {op_name} {key} {size} {version_id} {etag} +_JOURNAL_RECORD_RE = re.compile( + r'^(\S+)\s+(\S+)\s+\[([^\]]+)\]\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s*$' +) + + +def parse_journal_record(line): + m = _JOURNAL_RECORD_RE.match(line) + if not m: + return None + return { + 'bucket_owner': m.group(1), + 'bucket_name': m.group(2), + 'time': m.group(3), + 'op_name': m.group(4), + 'key': m.group(5), + 'size': m.group(6), + 'version_id': m.group(7), + 'etag': m.group(8), + } + + +def read_journal_records(s3_client, log_bucket, source_bucket, prefix=None, settle_time=5): + # settle_time waits for async log writes (async_completion=true) to drain + # before we flush; without it the pending log object may still be empty. + if prefix is None: + prefix = f'{source_bucket}/' + time.sleep(settle_time) + admin(['bucket', 'logging', 'flush', '--bucket', source_bucket]) + resp = s3_client.list_objects_v2(Bucket=log_bucket, Prefix=prefix) + keys = sorted(obj['Key'] for obj in resp.get('Contents', [])) + records = [] + for key in keys: + body = s3_client.get_object(Bucket=log_bucket, Key=key)['Body'].read().decode('utf-8') + for line in body.splitlines(): + parsed = parse_journal_record(line) + if parsed is not None: + records.append(parsed) + return records + + +@pytest.fixture +def lc_fast(): + set_lc_debug_interval(5) + yield 5 + + ##################### # bucket logging tests ##################### @@ -736,4 +850,185 @@ def test_logging_commands_unconfigured_bucket(s3_client): assert not output.strip(), f"{cmd} on unconfigured bucket should produce empty stdout: {output}" finally: - cleanup_bucket(s3_client, bucket) \ No newline at end of file + cleanup_bucket(s3_client, bucket) + + +@pytest.mark.basic_test +def test_lc_expiration_logs_journal_record(s3_client, logging_type, lc_fast): + """LC Expiration on a non-versioned bucket emits LIFECYCLE.DELETE.OBJECT""" + if logging_type != 'Journal': + pytest.skip("LC bucket logging is Journal-mode-only") + source = gen_bucket_name("lc-source") + log_bucket = gen_bucket_name("lc-log") + + try: + assert create_bucket_with_logging(s3_client, source, log_bucket, 'Journal') + apply_lc_config(s3_client, source, [make_lc_rule(Expiration={'Days': 1})]) + s3_client.put_object(Bucket=source, Key='obj.txt', Body=b'data') + + time.sleep(lc_fast + 2) + trigger_lc_processing() + assert wait_for_object_gone(s3_client, source, 'obj.txt'), "LC did not delete the object" + + records = read_journal_records(s3_client, log_bucket, source) + lc_records = [r for r in records if r['op_name'] == 'LIFECYCLE.DELETE.OBJECT'] + assert len(lc_records) == 1, f"expected 1 LIFECYCLE.DELETE.OBJECT record, got {len(lc_records)}: {records}" + assert lc_records[0]['key'] == 'obj.txt' + assert lc_records[0]['bucket_name'] == source + + finally: + cleanup_bucket(s3_client, source) + cleanup_bucket(s3_client, log_bucket) + + +@pytest.mark.basic_test +def test_lc_versioned_current_expiration_logs_journal_record(s3_client, logging_type, lc_fast): + """LC Expiration on a versioned bucket creates a delete marker and emits LIFECYCLE.DELETE.OBJECT for the expired current version.""" + if logging_type != 'Journal': + pytest.skip("LC bucket logging is Journal-mode-only") + source = gen_bucket_name("lc-source") + log_bucket = gen_bucket_name("lc-log") + + try: + assert create_bucket_with_logging(s3_client, source, log_bucket, 'Journal') + enable_versioning(s3_client, source) + + apply_lc_config(s3_client, source, [make_lc_rule(Expiration={'Days': 1})]) + v1 = s3_client.put_object(Bucket=source, Key='obj.txt', Body=b'data')['VersionId'] + + time.sleep(lc_fast + 2) + trigger_lc_processing() + time.sleep(lc_fast) + + resp = s3_client.list_object_versions(Bucket=source, Prefix='obj.txt') + versions = [v['VersionId'] for v in resp.get('Versions', [])] + delete_markers = resp.get('DeleteMarkers', []) + assert versions == [v1], f"expected v1 to remain as noncurrent, got {versions}" + assert len(delete_markers) == 1, f"expected 1 delete marker, got {len(delete_markers)}" + + records = read_journal_records(s3_client, log_bucket, source) + lc_records = [r for r in records if r['op_name'] == 'LIFECYCLE.DELETE.OBJECT'] + assert len(lc_records) == 1, f"expected 1 LIFECYCLE.DELETE.OBJECT record, got {len(lc_records)}" + assert lc_records[0]['version_id'] == v1 + assert lc_records[0]['key'] == 'obj.txt' + assert lc_records[0]['bucket_name'] == source + + finally: + cleanup_versioned_bucket(s3_client, source) + cleanup_bucket(s3_client, log_bucket) + + +@pytest.mark.basic_test +def test_lc_noncurrent_expiration_logs_journal_record(s3_client, logging_type, lc_fast): + """LC NoncurrentVersionExpiration emits LIFECYCLE.DELETE.OBJECT for the deleted noncurrent version.""" + if logging_type != 'Journal': + pytest.skip("LC bucket logging is Journal-mode-only") + source = gen_bucket_name("lc-source") + log_bucket = gen_bucket_name("lc-log") + + try: + assert create_bucket_with_logging(s3_client, source, log_bucket, 'Journal') + enable_versioning(s3_client, source) + + apply_lc_config(s3_client, source, [make_lc_rule(NoncurrentVersionExpiration={'NoncurrentDays': 1})]) + v1 = s3_client.put_object(Bucket=source, Key='obj.txt', Body=b'v1')['VersionId'] + v2 = s3_client.put_object(Bucket=source, Key='obj.txt', Body=b'v2')['VersionId'] # v1 -> noncurrent + + time.sleep(lc_fast + 2) + trigger_lc_processing() + time.sleep(lc_fast) + + resp = s3_client.list_object_versions(Bucket=source, Prefix='obj.txt') + remaining = [v['VersionId'] for v in resp.get('Versions', [])] + assert remaining == [v2], f"expected only current v2 to remain, got {remaining}" + + records = read_journal_records(s3_client, log_bucket, source) + lc_records = [r for r in records if r['op_name'] == 'LIFECYCLE.DELETE.OBJECT'] + assert len(lc_records) == 1, f"expected 1 LIFECYCLE.DELETE.OBJECT record, got {len(lc_records)}" + assert lc_records[0]['version_id'] == v1 + assert lc_records[0]['key'] == 'obj.txt' + assert lc_records[0]['bucket_name'] == source + + finally: + cleanup_versioned_bucket(s3_client, source) + cleanup_bucket(s3_client, log_bucket) + + +@pytest.mark.basic_test +def test_lc_dm_expiration_does_not_log(s3_client, logging_type, lc_fast): + """LC ExpiredObjectDeleteMarker must not emit LIFECYCLE.DELETE.OBJECT records.""" + if logging_type != 'Journal': + pytest.skip("LC bucket logging is Journal-mode-only") + source = gen_bucket_name("lc-source") + log_bucket = gen_bucket_name("lc-log") + + try: + assert create_bucket_with_logging(s3_client, source, log_bucket, 'Journal') + enable_versioning(s3_client, source) + + create_orphan_delete_marker(s3_client, source, 'obj.txt') + apply_lc_config(s3_client, source, [make_lc_rule(Expiration={'ExpiredObjectDeleteMarker': True})]) + + time.sleep(lc_fast + 2) + trigger_lc_processing() + time.sleep(lc_fast) + + resp = s3_client.list_object_versions(Bucket=source, Prefix='obj.txt') + assert not resp.get('Versions') and not resp.get('DeleteMarkers'), "delete marker was not expired, the rule never fired" + records = read_journal_records(s3_client, log_bucket, source) + lc_records = [r for r in records if r['op_name'] == 'LIFECYCLE.DELETE.OBJECT'] + assert len(lc_records) == 0, f"expected 0 LIFECYCLE.DELETE.OBJECT records, got {len(lc_records)}" + + finally: + cleanup_versioned_bucket(s3_client, source) + cleanup_bucket(s3_client, log_bucket) + + +@pytest.mark.basic_test +def test_lc_current_expiration_dm_branch_does_not_log(s3_client, logging_type, lc_fast): + """LC Expiration's delete-marker branch must not emit LIFECYCLE.DELETE.OBJECT records.""" + if logging_type != 'Journal': + pytest.skip("LC bucket logging is Journal-mode-only") + source = gen_bucket_name("lc-source") + log_bucket = gen_bucket_name("lc-log") + + try: + assert create_bucket_with_logging(s3_client, source, log_bucket, 'Journal') + enable_versioning(s3_client, source) + + create_orphan_delete_marker(s3_client, source, 'obj.txt') + apply_lc_config(s3_client, source, [make_lc_rule(Expiration={'Days': 1})]) + + time.sleep(lc_fast + 2) + trigger_lc_processing() + time.sleep(lc_fast) + + resp = s3_client.list_object_versions(Bucket=source, Prefix='obj.txt') + assert not resp.get('Versions') and not resp.get('DeleteMarkers'), "delete marker was not expired -- the rule never fired" + + records = read_journal_records(s3_client, log_bucket, source) + lc_records = [r for r in records if r['op_name'] == 'LIFECYCLE.DELETE.OBJECT'] + assert len(lc_records) == 0, f"expected 0 LIFECYCLE.DELETE.OBJECT records, got {len(lc_records)}" + + finally: + cleanup_versioned_bucket(s3_client, source) + cleanup_bucket(s3_client, log_bucket) + + +@pytest.mark.basic_test +def test_lc_runs_safely_without_logging_config(s3_client, lc_fast): + """LC runs successfully on a bucket without bucket-logging configured.""" + source = gen_bucket_name("lc-nolog") + + try: + s3_client.create_bucket(Bucket=source) + apply_lc_config(s3_client, source, [make_lc_rule(Expiration={'Days': 1})]) + s3_client.put_object(Bucket=source, Key='obj.txt', Body=b'data') + + time.sleep(lc_fast + 2) + _, ret = trigger_lc_processing() + assert ret == 0, f"radosgw-admin lc process failed: rc={ret}" + assert wait_for_object_gone(s3_client, source, 'obj.txt'), "LC did not delete the object" + + finally: + cleanup_bucket(s3_client, source) \ No newline at end of file