LU-12127 lfsck: skip orphan processing for namespace LFSCK

LFSCK can reconnect a recently-deleted orphan object back
into the normal namespace when it shouldn't. This can cause
access to the deleted data (potential security risk), and
sometimes cause an assertion if orphan is later deleted.

The commit 077570483e75e0610fd45149b926097547c434b8 skips
the orphan object during LFSCK scan. But what needs to be
skipped is only the namespace LFSCK logic for the orphan
object, but the layout LFSCK processing is still necessary
for the orphan object to verify the consistency between the
orphan MDT-object and related OST-object(s). This patch
just does that.

Signed-off-by: Fan Yong <fan.yong@intel.com>
Change-Id: I9b2809b95efa4b3c3e3b2c7d0a501624ed3ebbe5
Reviewed-on: https://review.whamcloud.com/34174
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Hongchao Zhang <hongchao@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/lfsck/lfsck_engine.c b/lustre/lfsck/lfsck_engine.c
index 606ea88..a1763a6 100644 (file)
--- a/lustre/lfsck/lfsck_engine.c
+++ b/lustre/lfsck/lfsck_engine.c
@@ -136,6 +136,7 @@ static int lfsck_needs_scan_dir(const struct lu_env *env,
        struct lfsck_thread_info *info    = lfsck_env_info(env);
        struct lu_fid            *fid     = &info->lti_fid;
        struct lu_seq_range      *range   = &info->lti_range;
+       struct lu_attr           *la      = &info->lti_la;
        struct seq_server_site   *ss      = lfsck_dev_site(lfsck);
        __u32                     idx     = lfsck_dev_idx(lfsck);
        int                       depth   = 0;
@@ -144,9 +145,21 @@ static int lfsck_needs_scan_dir(const struct lu_env *env,
        if (list_empty(&lfsck->li_list_dir) || !S_ISDIR(lfsck_object_type(obj)))
                return 0;
 
+       *fid = *lfsck_dto2fid(obj);
+       rc = dt_attr_get(env, obj, la);
+       if (unlikely(rc || (la->la_valid & LA_FLAGS &&
+                           la->la_flags & LUSTRE_ORPHAN_FL))) {
+               /* Orphan directory is empty, does not need scan. */
+               CDEBUG(D_INFO,
+                      "%s: skip orphan dir "DFID", %llx/%x: rc = %d\n",
+                      lfsck_lfsck2name(lfsck), PFID(fid),
+                      la->la_valid, la->la_flags, rc);
+
+               return rc;
+       }
+
        LASSERT(ss != NULL);
 
-       *fid = *lfsck_dto2fid(obj);
        while (1) {
                /* Global /ROOT is visible. */
                if (unlikely(lu_fid_eq(fid, &lfsck->li_global_root_fid)))
@@ -177,6 +190,7 @@ static int lfsck_needs_scan_dir(const struct lu_env *env,
                if (fid_is_norm(fid))
                        return 1;
 
+               /* Only true after "obj = NULL" set below */
                if (obj == NULL) {
                        obj = lfsck_object_find_bottom(env, lfsck, fid);
                        if (IS_ERR(obj))
@@ -960,19 +974,8 @@ static int lfsck_master_oit_engine(const struct lu_env *env,
                                goto checkpoint;
                }
 
-               if (dt_object_exists(target)) {
-                       struct lu_attr la = { .la_valid = 0 };
-
-                       rc = dt_attr_get(env, target, &la);
-                       if (likely(!rc && (!(la.la_valid & LA_FLAGS) ||
-                                          !(la.la_flags & LUSTRE_ORPHAN_FL))))
-                               rc = lfsck_exec_oit(env, lfsck, target);
-                       else
-                               CDEBUG(D_INFO,
-                                      "%s: orphan "DFID", %llx/%x: rc = %d\n",
-                                      lfsck_lfsck2name(lfsck), PFID(fid),
-                                      la.la_valid, la.la_flags, rc);
-               }
+               if (dt_object_exists(target))
+                       rc = lfsck_exec_oit(env, lfsck, target);
 
                lfsck_object_put(env, target);
                if (rc != 0 && bk->lb_param & LPF_FAILOUT)

diff --git a/lustre/lfsck/lfsck_namespace.c b/lustre/lfsck/lfsck_namespace.c
index 9606a11..973ca5c 100644 (file)
--- a/lustre/lfsck/lfsck_namespace.c
+++ b/lustre/lfsck/lfsck_namespace.c
@@ -4538,20 +4538,32 @@ static int lfsck_namespace_exec_oit(const struct lu_env *env,
                                    struct lfsck_component *com,
                                    struct dt_object *obj)
 {
-       struct lfsck_thread_info *info  = lfsck_env_info(env);
-       struct lfsck_namespace   *ns    = com->lc_file_ram;
-       struct lfsck_instance    *lfsck = com->lc_lfsck;
-       const struct lu_fid      *fid   = lfsck_dto2fid(obj);
-       struct lu_fid            *pfid  = &info->lti_fid2;
-       struct lu_name           *cname = &info->lti_name;
-       struct lu_seq_range      *range = &info->lti_range;
-       struct seq_server_site   *ss    = lfsck_dev_site(lfsck);
-       struct linkea_data        ldata = { NULL };
-       __u32                     idx   = lfsck_dev_idx(lfsck);
-       int                       rc;
+       struct lfsck_thread_info *info = lfsck_env_info(env);
+       struct lfsck_namespace *ns = com->lc_file_ram;
+       struct lfsck_instance *lfsck = com->lc_lfsck;
+       const struct lu_fid *fid = lfsck_dto2fid(obj);
+       struct lu_fid *pfid = &info->lti_fid2;
+       struct lu_name *cname = &info->lti_name;
+       struct lu_seq_range *range = &info->lti_range;
+       struct seq_server_site *ss = lfsck_dev_site(lfsck);
+       struct linkea_data ldata = { NULL };
+       __u32 idx = lfsck_dev_idx(lfsck);
+       struct lu_attr la = { .la_valid = 0 };
        bool remote = false;
+       int rc;
        ENTRY;
 
+       rc = dt_attr_get(env, obj, &la);
+       if (unlikely(rc || (la.la_valid & LA_FLAGS &&
+                           la.la_flags & LUSTRE_ORPHAN_FL))) {
+               CDEBUG(D_INFO,
+                      "%s: skip orphan "DFID", %llx/%x: rc = %d\n",
+                      lfsck_lfsck2name(lfsck), PFID(fid),
+                      la.la_valid, la.la_flags, rc);
+
+               return rc;
+       }
+
        rc = lfsck_links_read(env, obj, &ldata);
        if (rc == -ENOENT)
                GOTO(out, rc = 0);