gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-20 11:32:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(storage): integrate S3Operation into OperationHelper for unified metrics and audit (#2103)

Browse Source
This commit is contained in:
houseme
2026-03-08 17:57:33 +08:00
committed by GitHub
parent 8e4a1ef917
commit 60aa47bf61
39 changed files with 574 additions and 357 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+26 -26
View File
@@ -28,7 +28,6 @@ use crate::client::object_api_utils::new_getobjectreader;
use crate::error::Error;
use crate::error::StorageError;
use crate::error::{error_resp_to_object_err, is_err_object_not_found, is_err_version_not_found, is_network_or_host_down};
use crate::event::name::EventName;
use crate::event_notification::{EventArgs, send_event};
use crate::global::GLOBAL_LocalNodeName;
use crate::global::{GLOBAL_LifecycleSys, GLOBAL_TierConfigMgr, get_global_deployment_id};
@@ -45,6 +44,7 @@ use rustfs_common::data_usage::TierStats;
use rustfs_common::heal_channel::rep_has_active_rules;
use rustfs_common::metrics::{IlmAction, Metrics};
use rustfs_filemeta::{NULL_VERSION_ID, RestoreStatusOps, is_restored_object_on_disk};
use rustfs_s3_common::EventName;
use rustfs_utils::path::encode_dir_object;
use rustfs_utils::string::strings_has_prefix_fold;
use s3s::Body;
@@ -471,7 +471,7 @@ impl TransitionState {
}
pub async fn init(api: Arc<ECStore>) {
let max_workers = std::env::var("RUSTFS_MAX_TRANSITION_WORKERS")
let max_workers = env::var("RUSTFS_MAX_TRANSITION_WORKERS")
.ok()
.and_then(|s| s.parse::<i64>().ok())
.unwrap_or_else(|| std::cmp::min(num_cpus::get() as i64, 16));
@@ -569,14 +569,14 @@ impl TransitionState {
pub async fn update_workers_inner(api: Arc<ECStore>, n: i64) {
let mut n = n;
if n == 0 {
let max_workers = std::env::var("RUSTFS_MAX_TRANSITION_WORKERS")
let max_workers = env::var("RUSTFS_MAX_TRANSITION_WORKERS")
.ok()
.and_then(|s| s.parse::<i64>().ok())
.unwrap_or_else(|| std::cmp::min(num_cpus::get() as i64, 16));
n = max_workers;
}
// Allow environment override of maximum workers
let absolute_max = std::env::var("RUSTFS_ABSOLUTE_MAX_WORKERS")
let absolute_max = env::var("RUSTFS_ABSOLUTE_MAX_WORKERS")
.ok()
.and_then(|s| s.parse::<i64>().ok())
.unwrap_or(32);
@@ -603,7 +603,7 @@ impl TransitionState {
}
pub async fn init_background_expiry(api: Arc<ECStore>) {
let mut workers = std::env::var("RUSTFS_MAX_EXPIRY_WORKERS")
let mut workers = env::var("RUSTFS_MAX_EXPIRY_WORKERS")
.ok()
.and_then(|s| s.parse::<usize>().ok())
.unwrap_or_else(|| std::cmp::min(num_cpus::get(), 16));
@@ -615,7 +615,7 @@ pub async fn init_background_expiry(api: Arc<ECStore>) {
}
if workers == 0 {
workers = std::env::var("RUSTFS_DEFAULT_EXPIRY_WORKERS")
workers = env::var("RUSTFS_DEFAULT_EXPIRY_WORKERS")
.ok()
.and_then(|s| s.parse::<usize>().ok())
.unwrap_or(8);
@@ -689,13 +689,13 @@ pub async fn expire_transitioned_object(
//let tags = LcAuditEvent::new(src, lcEvent).Tags();
if lc_event.action == IlmAction::DeleteRestoredAction {
opts.transition.expire_restored = true;
match api.delete_object(&oi.bucket, &oi.name, opts).await {
return match api.delete_object(&oi.bucket, &oi.name, opts).await {
Ok(dobj) => {
//audit_log_lifecycle(*oi, ILMExpiry, tags, traceFn);
return Ok(dobj);
Ok(dobj)
}
Err(err) => return Err(std::io::Error::other(err)),
}
Err(err) => Err(std::io::Error::other(err)),
};
}
let ret = delete_object_from_remote_tier(
@@ -732,7 +732,7 @@ pub async fn expire_transitioned_object(
..Default::default()
};
send_event(EventArgs {
event_name: event_name.as_ref().to_string(),
event_name: event_name.to_string(),
bucket_name: obj_info.bucket.clone(),
object: obj_info,
user_agent: "Internal: [ILM-Expiry]".to_string(),
@@ -847,8 +847,8 @@ pub async fn post_restore_opts(version_id: &str, bucket: &str, object: &str) ->
}
}
Ok(ObjectOptions {
versioned: versioned,
version_suspended: version_suspended,
versioned,
version_suspended,
version_id: Some(vid.to_string()),
..Default::default()
})
@@ -1033,12 +1033,12 @@ pub async fn eval_action_from_lifecycle(
let lock_enabled = if let Some(lr) = lr { lr.mode.is_some() } else { false };
match event.action {
lifecycle::IlmAction::DeleteAllVersionsAction | lifecycle::IlmAction::DelMarkerDeleteAllVersionsAction => {
IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => {
if lock_enabled {
return lifecycle::Event::default();
}
}
lifecycle::IlmAction::DeleteVersionAction | lifecycle::IlmAction::DeleteRestoredVersionAction => {
IlmAction::DeleteVersionAction | IlmAction::DeleteRestoredVersionAction => {
if oi.version_id.is_none() {
return lifecycle::Event::default();
}
@@ -1139,12 +1139,12 @@ pub async fn apply_expiry_on_non_transitioned_objects(
event_name = EventName::ObjectRemovedDeleteMarkerCreated;
}
match lc_event.action {
lifecycle::IlmAction::DeleteAllVersionsAction => event_name = EventName::ObjectRemovedDeleteAllVersions,
lifecycle::IlmAction::DelMarkerDeleteAllVersionsAction => event_name = EventName::ILMDelMarkerExpirationDelete,
IlmAction::DeleteAllVersionsAction => event_name = EventName::ObjectRemovedDeleteAllVersions,
IlmAction::DelMarkerDeleteAllVersionsAction => event_name = EventName::LifecycleDelMarkerExpirationDelete,
_ => (),
}
send_event(EventArgs {
event_name: event_name.as_ref().to_string(),
event_name: event_name.to_string(),
bucket_name: dobj.bucket.clone(),
object: dobj,
user_agent: "Internal: [ILM-Expiry]".to_string(),
@@ -1152,7 +1152,7 @@ pub async fn apply_expiry_on_non_transitioned_objects(
..Default::default()
});
if lc_event.action != lifecycle::IlmAction::NoneAction {
if lc_event.action != IlmAction::NoneAction {
let mut num_versions = 1_u64;
if lc_event.action.delete_all() {
num_versions = oi.num_versions as u64;
@@ -1172,15 +1172,15 @@ pub async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &
pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
let mut success = false;
match event.action {
lifecycle::IlmAction::DeleteVersionAction
| lifecycle::IlmAction::DeleteAction
| lifecycle::IlmAction::DeleteRestoredAction
| lifecycle::IlmAction::DeleteRestoredVersionAction
| lifecycle::IlmAction::DeleteAllVersionsAction
| lifecycle::IlmAction::DelMarkerDeleteAllVersionsAction => {
IlmAction::DeleteVersionAction
| IlmAction::DeleteAction
| IlmAction::DeleteRestoredAction
| IlmAction::DeleteRestoredVersionAction
| IlmAction::DeleteAllVersionsAction
| IlmAction::DelMarkerDeleteAllVersionsAction => {
success = apply_expiry_rule(event, src, oi).await;
}
lifecycle::IlmAction::TransitionAction | lifecycle::IlmAction::TransitionVersionAction => {
IlmAction::TransitionAction | IlmAction::TransitionVersionAction => {
success = apply_transition_rule(event, src, oi).await;
}
_ => (),
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+66 -63
View File
@@ -26,7 +26,6 @@ use crate::client::api_get_options::{AdvancedGetOptions, StatObjectOptions};
use crate::config::com::save_config;
use crate::disk::BUCKET_META_PREFIX;
use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_found};
use crate::event::name::EventName;
use crate::event_notification::{EventArgs, send_event};
use crate::global::GLOBAL_LocalNodeName;
use crate::global::get_global_bucket_monitor;
@@ -41,6 +40,10 @@ use aws_smithy_types::body::SdkBody;
use byteorder::ByteOrder;
use futures::future::join_all;
use futures::stream::StreamExt;
use headers::{
AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, AMZ_SERVER_SIDE_ENCRYPTION,
AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_LANGUAGE, CONTENT_TYPE,
};
use http::HeaderMap;
use http_body::Frame;
use http_body_util::StreamBody;
@@ -51,6 +54,7 @@ use rustfs_filemeta::{
ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType,
get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map,
};
use rustfs_s3_common::EventName;
use rustfs_utils::http::{
AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, AMZ_TAGGING_DIRECTIVE, CONTENT_ENCODING, HeaderExt as _,
RESERVED_METADATA_PREFIX, RESERVED_METADATA_PREFIX_LOWER, RUSTFS_REPLICATION_ACTUAL_OBJECT_SIZE,
@@ -570,7 +574,7 @@ impl ReplicationResyncer {
return;
}
let worker_idx = sip_hash(&roi.name, RESYNC_WORKER_COUNT, &DEFAULT_SIP_HASH_KEY) as usize;
let worker_idx = sip_hash(&roi.name, RESYNC_WORKER_COUNT, &DEFAULT_SIP_HASH_KEY);
if let Err(err) = worker_txs[worker_idx].send(roi).await {
error!("Failed to send object info to worker: {}", err);
@@ -1087,7 +1091,7 @@ pub async fn check_replicate_delete(
}
/// Check if the user-defined metadata contains SSEC encryption headers
fn is_ssec_encrypted(user_defined: &std::collections::HashMap<String, String>) -> bool {
fn is_ssec_encrypted(user_defined: &HashMap<String, String>) -> bool {
user_defined.contains_key(SSEC_ALGORITHM_HEADER)
|| user_defined.contains_key(SSEC_KEY_HEADER)
|| user_defined.contains_key(SSEC_KEY_MD5_HEADER)
@@ -1224,7 +1228,7 @@ pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo,
Ok(None) => {
warn!("No replication config found for bucket: {}", bucket);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1243,7 +1247,7 @@ pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo,
Err(err) => {
warn!("replication config for bucket: {} error: {}", bucket, err);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1276,7 +1280,7 @@ pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo,
bucket, dobj.target_arn, err
);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1304,7 +1308,7 @@ pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo,
bucket, dobj.delete_object.object_name, e
);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1329,7 +1333,7 @@ pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo,
bucket, dobj.delete_object.object_name, e
);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1370,7 +1374,7 @@ pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo,
let Some(tgt_client) = BucketTargetSys::get().get_remote_target_client(&bucket, &tgt_entry.arn).await else {
warn!("failed to get target for bucket:{:?}, arn:{:?}", &bucket, &tgt_entry.arn);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1401,7 +1405,7 @@ pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo,
Err(e) => {
error!("replicate_delete task failed: {}", e);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1454,9 +1458,9 @@ pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo,
}
let event_name = if replication_status == ReplicationStatusType::Completed {
EventName::ObjectReplicationComplete.as_ref().to_string()
EventName::ObjectReplicationComplete.to_string()
} else {
EventName::ObjectReplicationFailed.as_ref().to_string()
EventName::ObjectReplicationFailed.to_string()
};
match storage
@@ -1509,7 +1513,7 @@ async fn replicate_force_delete_to_targets<S: StorageAPI>(dobj: &DeletedObjectRe
Ok(None) => {
warn!("replicate force-delete: no replication config for bucket:{}", bucket);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1525,7 +1529,7 @@ async fn replicate_force_delete_to_targets<S: StorageAPI>(dobj: &DeletedObjectRe
Err(err) => {
warn!("replicate force-delete: replication config error bucket:{} error:{}", bucket, err);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1551,7 +1555,7 @@ async fn replicate_force_delete_to_targets<S: StorageAPI>(dobj: &DeletedObjectRe
bucket, object_name, e
);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1574,7 +1578,7 @@ async fn replicate_force_delete_to_targets<S: StorageAPI>(dobj: &DeletedObjectRe
bucket, object_name, e
);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1604,7 +1608,7 @@ async fn replicate_force_delete_to_targets<S: StorageAPI>(dobj: &DeletedObjectRe
let Some(tgt_client) = BucketTargetSys::get().get_remote_target_client(bucket, &arn).await else {
warn!("replicate force-delete: failed to get target client bucket:{} arn:{}", bucket, arn);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1625,7 +1629,7 @@ async fn replicate_force_delete_to_targets<S: StorageAPI>(dobj: &DeletedObjectRe
if BucketTargetSys::get().is_offline(&tgt_client.to_url()).await {
error!("replicate force-delete: target offline bucket:{} arn:{}", bucket, tgt_client.arn);
send_event(EventArgs {
event_name: EventName::ObjectReplicationFailed.as_ref().to_string(),
event_name: EventName::ObjectReplicationFailed.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1661,7 +1665,7 @@ async fn replicate_force_delete_to_targets<S: StorageAPI>(dobj: &DeletedObjectRe
bucket, object_name, tgt_client.arn, e
);
send_event(EventArgs {
event_name: EventName::ObjectReplicationFailed.as_ref().to_string(),
event_name: EventName::ObjectReplicationFailed.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
@@ -1794,7 +1798,7 @@ pub async fn replicate_object<S: StorageAPI>(roi: ReplicateObjectInfo, storage:
Ok(None) => {
warn!("No replication config found for bucket: {}", bucket);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: GLOBAL_LocalNodeName.to_string(),
@@ -1806,7 +1810,7 @@ pub async fn replicate_object<S: StorageAPI>(roi: ReplicateObjectInfo, storage:
Err(err) => {
error!("Failed to get replication config for bucket {}: {}", bucket, err);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: GLOBAL_LocalNodeName.to_string(),
@@ -1832,7 +1836,7 @@ pub async fn replicate_object<S: StorageAPI>(roi: ReplicateObjectInfo, storage:
let Some(tgt_client) = BucketTargetSys::get().get_remote_target_client(&bucket, &arn).await else {
warn!("failed to get target for bucket:{:?}, arn:{:?}", &bucket, &arn);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: GLOBAL_LocalNodeName.to_string(),
@@ -1866,7 +1870,7 @@ pub async fn replicate_object<S: StorageAPI>(roi: ReplicateObjectInfo, storage:
Err(e) => {
error!("replicate_object task failed: {}", e);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: GLOBAL_LocalNodeName.to_string(),
@@ -1900,9 +1904,9 @@ pub async fn replicate_object<S: StorageAPI>(roi: ReplicateObjectInfo, storage:
}
let event_name = if replication_status == ReplicationStatusType::Completed {
EventName::ObjectReplicationComplete.as_ref().to_string()
EventName::ObjectReplicationComplete.to_string()
} else {
EventName::ObjectReplicationFailed.as_ref().to_string()
EventName::ObjectReplicationFailed.to_string()
};
send_event(EventArgs {
@@ -1957,7 +1961,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
if BucketTargetSys::get().is_offline(&tgt_client.to_url()).await {
warn!("target is offline: {}", tgt_client.to_url());
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: self.to_object_info(),
host: GLOBAL_LocalNodeName.to_string(),
@@ -1988,7 +1992,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
warn!("failed to get object reader for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: self.to_object_info(),
host: GLOBAL_LocalNodeName.to_string(),
@@ -2010,7 +2014,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
Err(e) => {
warn!("failed to get actual size for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: GLOBAL_LocalNodeName.to_string(),
@@ -2024,7 +2028,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
if tgt_client.bucket.is_empty() {
warn!("target bucket is empty: {}", tgt_client.bucket);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: GLOBAL_LocalNodeName.to_string(),
@@ -2081,7 +2085,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
bucket, tgt_client.arn, e
);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: GLOBAL_LocalNodeName.to_string(),
@@ -2154,7 +2158,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
if BucketTargetSys::get().is_offline(&tgt_client.to_url()).await {
warn!("target is offline: {}", tgt_client.to_url());
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: self.to_object_info(),
host: GLOBAL_LocalNodeName.to_string(),
@@ -2184,7 +2188,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
if !is_err_object_not_found(&e) || is_err_version_not_found(&e) {
warn!("failed to get object reader for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: self.to_object_info(),
host: GLOBAL_LocalNodeName.to_string(),
@@ -2215,7 +2219,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
Err(e) => {
warn!("failed to get actual size for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: GLOBAL_LocalNodeName.to_string(),
@@ -2231,7 +2235,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
if tgt_client.bucket.is_empty() {
warn!("target bucket is empty: {}", tgt_client.bucket);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: GLOBAL_LocalNodeName.to_string(),
@@ -2262,9 +2266,8 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
if replication_action == ReplicationAction::None {
if self.op_type == ReplicationType::ExistingObject
&& object_info.mod_time
> oi.last_modified.map(|dt| {
time::OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(time::OffsetDateTime::UNIX_EPOCH)
})
> oi.last_modified
.map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH))
&& object_info.version_id.is_none()
{
warn!(
@@ -2274,7 +2277,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
tgt_client.to_url()
);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info.clone(),
host: GLOBAL_LocalNodeName.to_string(),
@@ -2314,7 +2317,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
warn!("failed to head object for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: GLOBAL_LocalNodeName.to_string(),
@@ -2330,7 +2333,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
warn!("failed to head object for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: GLOBAL_LocalNodeName.to_string(),
@@ -2360,7 +2363,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
bucket, tgt_client.arn, e
);
send_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(),
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: GLOBAL_LocalNodeName.to_string(),
@@ -2431,19 +2434,19 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
// Standard headers that needs to be extracted from User metadata.
static STANDARD_HEADERS: &[&str] = &[
headers::CONTENT_TYPE,
headers::CACHE_CONTROL,
headers::CONTENT_ENCODING,
headers::CONTENT_LANGUAGE,
headers::CONTENT_DISPOSITION,
headers::AMZ_STORAGE_CLASS,
headers::AMZ_OBJECT_TAGGING,
headers::AMZ_BUCKET_REPLICATION_STATUS,
headers::AMZ_OBJECT_LOCK_MODE,
headers::AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE,
headers::AMZ_OBJECT_LOCK_LEGAL_HOLD,
headers::AMZ_TAG_COUNT,
headers::AMZ_SERVER_SIDE_ENCRYPTION,
CONTENT_TYPE,
CACHE_CONTROL,
CONTENT_ENCODING,
CONTENT_LANGUAGE,
CONTENT_DISPOSITION,
AMZ_STORAGE_CLASS,
AMZ_OBJECT_TAGGING,
AMZ_BUCKET_REPLICATION_STATUS,
AMZ_OBJECT_LOCK_MODE,
AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE,
AMZ_OBJECT_LOCK_LEGAL_HOLD,
AMZ_TAG_COUNT,
AMZ_SERVER_SIDE_ENCRYPTION,
];
fn calc_put_object_header_size(put_opts: &PutObjectOptions) -> usize {
@@ -2577,7 +2580,7 @@ fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObject
meta.insert(REPLICATION_SSEC_CHECKSUM_HEADER.to_string(), encoded);
} else {
// Get checksum metadata for non-SSE-C objects
let (cs_meta, is_mp) = object_info.decrypt_checksums(0, &http::HeaderMap::new())?;
let (cs_meta, is_mp) = object_info.decrypt_checksums(0, &HeaderMap::new())?;
is_multipart = is_mp;
// Set object checksum metadata
@@ -2649,24 +2652,24 @@ fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObject
// Use case-insensitive lookup for headers
let lk_map = object_info.user_defined.clone();
if let Some(lang) = lk_map.lookup(headers::CONTENT_LANGUAGE) {
if let Some(lang) = lk_map.lookup(CONTENT_LANGUAGE) {
put_op.content_language = lang.to_string();
}
if let Some(cd) = lk_map.lookup(headers::CONTENT_DISPOSITION) {
if let Some(cd) = lk_map.lookup(CONTENT_DISPOSITION) {
put_op.content_disposition = cd.to_string();
}
if let Some(v) = lk_map.lookup(headers::CACHE_CONTROL) {
if let Some(v) = lk_map.lookup(CACHE_CONTROL) {
put_op.cache_control = v.to_string();
}
if let Some(v) = lk_map.lookup(headers::AMZ_OBJECT_LOCK_MODE) {
if let Some(v) = lk_map.lookup(AMZ_OBJECT_LOCK_MODE) {
let mode = v.to_string().to_uppercase();
put_op.mode = Some(aws_sdk_s3::types::ObjectLockRetentionMode::from(mode.as_str()));
}
if let Some(v) = lk_map.lookup(headers::AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) {
if let Some(v) = lk_map.lookup(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) {
put_op.retain_until_date =
OffsetDateTime::parse(v, &Rfc3339).map_err(|e| Error::other(format!("Failed to parse retain until date: {}", e)))?;
// set retention timestamp in opts
@@ -2680,7 +2683,7 @@ fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObject
};
}
if let Some(v) = lk_map.lookup(headers::AMZ_OBJECT_LOCK_LEGAL_HOLD) {
if let Some(v) = lk_map.lookup(AMZ_OBJECT_LOCK_LEGAL_HOLD) {
let hold = v.to_uppercase();
put_op.legalhold = Some(ObjectLockLegalHoldStatus::from(hold.as_str()));
// set legalhold timestamp in opts
@@ -2865,7 +2868,7 @@ fn get_replication_action(oi1: &ObjectInfo, oi2: &HeadObjectOutput, op_type: Rep
&& oi1.mod_time
> oi2
.last_modified
.map(|dt| time::OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(time::OffsetDateTime::UNIX_EPOCH))
.map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH))
&& oi1.version_id.is_none()
{
return ReplicationAction::None;
@@ -2884,7 +2887,7 @@ fn get_replication_action(oi1: &ObjectInfo, oi2: &HeadObjectOutput, op_type: Rep
|| oi1.mod_time
!= oi2
.last_modified
.map(|dt| time::OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(time::OffsetDateTime::UNIX_EPOCH))
.map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH))
{
return ReplicationAction::All;
}