feat(lifecycle): Implement object lifecycle management functionality (#358)

* feat(lifecycle): Implement object lifecycle management functionality Add a lifecycle module to automatically handle object expiration and transition during scanning Modify the file metadata cache module to be publicly visible to support lifecycle operations Adjust the scanning interval to a shorter time for testing lifecycle rules Implement the parsing and execution logic for S3 lifecycle configurations Add integration tests to verify the lifecycle expiration functionality Update dependencies to support the new lifecycle features Signed-off-by: junxiang Mu <1948535941@qq.com> * fix cargo dependencies Signed-off-by: junxiang Mu <1948535941@qq.com> * fix fmt Signed-off-by: junxiang Mu <1948535941@qq.com> --------- Signed-off-by: junxiang Mu <1948535941@qq.com> Co-authored-by: houseme <housemecn@gmail.com>
2026-08-11 07:36:53 +00:00 · 2025-08-07 19:51:02 -07:00
parent 48a9707110
commit d987686c14
12 changed files with 756 additions and 111 deletions
@@ -516,7 +516,7 @@ impl TransitionState {
                        if let Err(err) = transition_object(api.clone(), &task.obj_info, LcAuditEvent::new(task.event.clone(), task.src.clone())).await {
                            if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") {
                                error!("Transition to {} failed for {}/{} version:{} with {}",
-                                    task.event.storage_class, task.obj_info.bucket, task.obj_info.name, task.obj_info.version_id.expect("err"), err.to_string());
+                                    task.event.storage_class, task.obj_info.bucket, task.obj_info.name, task.obj_info.version_id.map(|v| v.to_string()).unwrap_or_default(), err.to_string());
                            }
                        } else {
                            let mut ts = TierStats {
@@ -743,7 +743,7 @@ pub async fn transition_object(api: Arc<ECStore>, oi: &ObjectInfo, lae: LcAuditE
            ..Default::default()
        },
        //lifecycle_audit_event: lae,
-        version_id: Some(oi.version_id.expect("err").to_string()),
+        version_id: oi.version_id.map(|v| v.to_string()),
        versioned: BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await,
        version_suspended: BucketVersioningSys::prefix_suspended(&oi.bucket, &oi.name).await,
        mod_time: oi.mod_time,
@@ -808,7 +808,7 @@ impl LifecycleOps for ObjectInfo {
        lifecycle::ObjectOpts {
            name: self.name.clone(),
            user_tags: self.user_tags.clone(),
-            version_id: self.version_id.expect("err").to_string(),
+            version_id: self.version_id.map(|v| v.to_string()).unwrap_or_default(),
            mod_time: self.mod_time,
            size: self.size as usize,
            is_latest: self.is_latest,
@@ -874,7 +874,11 @@ pub async fn eval_action_from_lifecycle(
            if lock_enabled && enforce_retention_for_deletion(oi) {
                //if serverDebugLog {
                if oi.version_id.is_some() {
-                    info!("lifecycle: {} v({}) is locked, not deleting", oi.name, oi.version_id.expect("err"));
+                    info!(
+                        "lifecycle: {} v({}) is locked, not deleting",
+                        oi.name,
+                        oi.version_id.map(|v| v.to_string()).unwrap_or_default()
+                    );
                } else {
                    info!("lifecycle: {} is locked, not deleting", oi.name);
                }
@@ -928,7 +932,7 @@ pub async fn apply_expiry_on_non_transitioned_objects(
    };

    if lc_event.action.delete_versioned() {
-        opts.version_id = Some(oi.version_id.expect("err").to_string());
+        opts.version_id = oi.version_id.map(|v| v.to_string());
    }

    opts.versioned = BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await;
@@ -27,6 +27,7 @@ use std::env;
 use std::fmt::Display;
 use time::macros::{datetime, offset};
 use time::{self, Duration, OffsetDateTime};
+use tracing::info;

 use crate::bucket::lifecycle::rule::TransitionOps;

@@ -279,7 +280,12 @@ impl Lifecycle for BucketLifecycleConfiguration {

    async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime) -> Event {
        let mut events = Vec::<Event>::new();
+        info!(
+            "eval_inner: object={}, mod_time={:?}, now={:?}, is_latest={}, delete_marker={}",
+            obj.name, obj.mod_time, now, obj.is_latest, obj.delete_marker
+        );
        if obj.mod_time.expect("err").unix_timestamp() == 0 {
+            info!("eval_inner: mod_time is 0, returning default event");
            return Event::default();
        }

@@ -418,7 +424,16 @@ impl Lifecycle for BucketLifecycleConfiguration {
                    }
                }

-                if obj.is_latest && !obj.delete_marker {
+                info!(
+                    "eval_inner: checking expiration condition - is_latest={}, delete_marker={}, version_id={:?}, condition_met={}",
+                    obj.is_latest,
+                    obj.delete_marker,
+                    obj.version_id,
+                    (obj.is_latest || obj.version_id.is_empty()) && !obj.delete_marker
+                );
+                // Allow expiration for latest objects OR non-versioned objects (empty version_id)
+                if (obj.is_latest || obj.version_id.is_empty()) && !obj.delete_marker {
+                    info!("eval_inner: entering expiration check");
                    if let Some(ref expiration) = rule.expiration {
                        if let Some(ref date) = expiration.date {
                            let date0 = OffsetDateTime::from(date.clone());
@@ -435,22 +450,29 @@ impl Lifecycle for BucketLifecycleConfiguration {
                                });
                            }
                        } else if let Some(days) = expiration.days {
-                            if days != 0 {
-                                let expected_expiry: OffsetDateTime = expected_expiry_time(obj.mod_time.expect("err!"), days);
-                                if now.unix_timestamp() == 0 || now.unix_timestamp() > expected_expiry.unix_timestamp() {
-                                    let mut event = Event {
-                                        action: IlmAction::DeleteAction,
-                                        rule_id: rule.id.clone().expect("err!"),
-                                        due: Some(expected_expiry),
-                                        noncurrent_days: 0,
-                                        newer_noncurrent_versions: 0,
-                                        storage_class: "".into(),
-                                    };
-                                    /*if rule.expiration.expect("err!").delete_all.val {
-                                        event.action = IlmAction::DeleteAllVersionsAction
-                                    }*/
-                                    events.push(event);
-                                }
+                            let expected_expiry: OffsetDateTime = expected_expiry_time(obj.mod_time.expect("err!"), days);
+                            info!(
+                                "eval_inner: expiration check - days={}, obj_time={:?}, expiry_time={:?}, now={:?}, should_expire={}",
+                                days,
+                                obj.mod_time.expect("err!"),
+                                expected_expiry,
+                                now,
+                                now.unix_timestamp() > expected_expiry.unix_timestamp()
+                            );
+                            if now.unix_timestamp() == 0 || now.unix_timestamp() > expected_expiry.unix_timestamp() {
+                                info!("eval_inner: object should expire, adding DeleteAction");
+                                let mut event = Event {
+                                    action: IlmAction::DeleteAction,
+                                    rule_id: rule.id.clone().expect("err!"),
+                                    due: Some(expected_expiry),
+                                    noncurrent_days: 0,
+                                    newer_noncurrent_versions: 0,
+                                    storage_class: "".into(),
+                                };
+                                /*if rule.expiration.expect("err!").delete_all.val {
+                                    event.action = IlmAction::DeleteAllVersionsAction
+                                }*/
+                                events.push(event);
                            }
                        }
                    }
@@ -598,7 +620,7 @@ impl LifecycleCalculate for Transition {

 pub fn expected_expiry_time(mod_time: OffsetDateTime, days: i32) -> OffsetDateTime {
    if days == 0 {
-        return mod_time;
+        return OffsetDateTime::UNIX_EPOCH; // Return epoch time to ensure immediate expiry
    }
    let t = mod_time
        .to_offset(offset!(-0:00:00))
@@ -3837,7 +3837,90 @@ impl StorageAPI for SetDisks {

            return Ok(ObjectInfo::default());
        }
-        unimplemented!()
+
+        // Create a single object deletion request
+        let mut vr = FileInfo {
+            name: object.to_string(),
+            version_id: opts.version_id.as_ref().and_then(|v| Uuid::parse_str(v).ok()),
+            ..Default::default()
+        };
+
+        // Handle versioning
+        let (suspended, versioned) = (opts.version_suspended, opts.versioned);
+        if opts.version_id.is_none() && (suspended || versioned) {
+            vr.mod_time = Some(OffsetDateTime::now_utc());
+            vr.deleted = true;
+            if versioned {
+                vr.version_id = Some(Uuid::new_v4());
+            }
+        }
+
+        let vers = vec![FileInfoVersions {
+            name: vr.name.clone(),
+            versions: vec![vr.clone()],
+            ..Default::default()
+        }];
+
+        let disks = self.disks.read().await;
+        let disks = disks.clone();
+        let write_quorum = disks.len() / 2 + 1;
+
+        let mut futures = Vec::with_capacity(disks.len());
+        let mut errs = Vec::with_capacity(disks.len());
+
+        for disk in disks.iter() {
+            let vers = vers.clone();
+            futures.push(async move {
+                if let Some(disk) = disk {
+                    disk.delete_versions(bucket, vers, DeleteOptions::default()).await
+                } else {
+                    Err(DiskError::DiskNotFound)
+                }
+            });
+        }
+
+        let results = join_all(futures).await;
+
+        for result in results {
+            match result {
+                Ok(disk_errs) => {
+                    // Handle errors from disk operations
+                    for err in disk_errs.iter().flatten() {
+                        warn!("delete_object disk error: {:?}", err);
+                    }
+                    errs.push(None);
+                }
+                Err(e) => {
+                    errs.push(Some(e));
+                }
+            }
+        }
+
+        // Check write quorum
+        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+            return Err(to_object_err(err.into(), vec![bucket, object]));
+        }
+
+        // Create result ObjectInfo
+        let result_info = if vr.deleted {
+            ObjectInfo {
+                bucket: bucket.to_string(),
+                name: object.to_string(),
+                delete_marker: true,
+                mod_time: vr.mod_time,
+                version_id: vr.version_id,
+                ..Default::default()
+            }
+        } else {
+            ObjectInfo {
+                bucket: bucket.to_string(),
+                name: object.to_string(),
+                version_id: vr.version_id,
+                ..Default::default()
+            }
+        };
+
+        Ok(result_info)
    }

    #[tracing::instrument(skip(self))]