}
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;
#include <random>
#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"
return extracted_time;
}
-std::string full_bucket_name(const std::unique_ptr<rgw::sal::Bucket>& bucket) {
+std::string full_bucket_name(const rgw::sal::Bucket* bucket) {
if (bucket->get_tenant().empty()) {
return bucket->get_name();
}
std::string& obj_name,
const DoutPrefixProvider *dpp,
const std::string& region,
- const std::unique_ptr<rgw::sal::Bucket>& source_bucket,
+ rgw::sal::Bucket* source_bucket,
optional_yield y,
std::optional<std::string> old_name,
RGWObjVersionTracker* objv_tracker) {
std::string& obj_name,
const DoutPrefixProvider *dpp,
const std::string& region,
- const std::unique_ptr<rgw::sal::Bucket>& source_bucket,
+ rgw::sal::Bucket* source_bucket,
optional_yield y,
bool must_commit,
RGWObjVersionTracker* objv_tracker,
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<std::chrono::milliseconds>(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));
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;
}
}
// 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();
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;
}
}
} 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;
}
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) {
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<std::chrono::milliseconds>(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}] {} {} {} {} {}",
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:
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;
}
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;
}
}
"'. 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;
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:
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());
}
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) {
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,
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
std::string& obj_name,
const DoutPrefixProvider *dpp,
const std::string& region,
- const std::unique_ptr<rgw::sal::Bucket>& source_bucket,
+ rgw::sal::Bucket* source_bucket,
optional_yield y,
bool must_commit,
RGWObjVersionTracker* objv_tracker,
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:]<bucket name>
int get_bucket_id(const std::string& bucket_name, const std::string& tenant_name, rgw_bucket& bucket_id);
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
--- /dev/null
+// -*- 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 <chrono>
+#include <cstddef>
+#include <string>
+#include <string_view>
+#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
#include <boost/algorithm/string/split.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string/predicate.hpp>
+#include <boost/optional.hpp>
#include "include/scope_guard.h"
#include "include/function2.hpp"
#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"
}
}
+// 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)
{
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<const std::string&> log_op_name = boost::none) {
int ret{0};
auto& driver = oc.driver;
auto& bucket_info = oc.bucket->get_info();
}
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()) {
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;
/* ! 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
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
*/
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
obj_name,
this,
region,
- src_bucket,
+ src_bucket.get(),
y,
false, // rollover should happen even if commit failed
&objv_tracker,
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 << "'. "
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
#####################
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
projects / ceph.git / commitdiff