LU-13693 lfs: check early for MDS_OPEN_DIRECTORY

[fs/lustre-release.git] / lustre / mdt / mdt_open.c

diff --git a/lustre/mdt/mdt_open.c b/lustre/mdt/mdt_open.c
index bcb222a..5aa5314 100644 (file)
--- a/lustre/mdt/mdt_open.c
+++ b/lustre/mdt/mdt_open.c
@@ -1318,12 +1318,12 @@ int mdt_reint_open(struct mdt_thread_info *info, struct mdt_lock_handle *lhc)
        int created = 0;
        int object_locked = 0;
        u32 msg_flags;
+       ktime_t kstart = ktime_get();
 
        ENTRY;
        OBD_FAIL_TIMEOUT_ORSET(OBD_FAIL_MDS_PAUSE_OPEN, OBD_FAIL_ONCE,
                               (obd_timeout + 1) / 4);
 
-       mdt_counter_incr(req, LPROC_MDT_OPEN);
        repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 
        ma->ma_need = MA_INODE;
@@ -1497,7 +1497,8 @@ int mdt_reint_open(struct mdt_thread_info *info, struct mdt_lock_handle *lhc)
                                 GOTO(out_child, result);
                 }
                created = 1;
-               mdt_counter_incr(req, LPROC_MDT_MKNOD);
+               mdt_counter_incr(req, LPROC_MDT_MKNOD,
+                                ktime_us_delta(ktime_get(), kstart));
         } else {
                 /*
                  * The object is on remote node, return its FID for remote open.
@@ -1528,6 +1529,14 @@ int mdt_reint_open(struct mdt_thread_info *info, struct mdt_lock_handle *lhc)
                                result = -MDT_EREMOTE_OPEN;
                         GOTO(out_child, result);
                } else if (mdt_object_exists(child)) {
+                       /* Check early for MDS_OPEN_DIRECTORY/O_DIRECTORY to
+                        * avoid opening regular files from lfs getstripe
+                        * since doing so breaks the leases used by lfs
+                        * mirror. See LU-13693. */
+                       if (open_flags & MDS_OPEN_DIRECTORY &&
+                           S_ISREG(lu_object_attr(&child->mot_obj)))
+                               GOTO(out_child, result = -ENOTDIR);
+
                        /* We have to get attr & LOV EA & HSM for this
                         * object. */
                        mdt_prep_ma_buf_from_rep(info, child, ma);
@@ -1603,6 +1612,10 @@ int mdt_reint_open(struct mdt_thread_info *info, struct mdt_lock_handle *lhc)
                        mdt_clear_disposition(info, ldlm_rep, DISP_OPEN_CREATE);
                }
        }
+
+       mdt_counter_incr(req, LPROC_MDT_OPEN,
+                        ktime_us_delta(ktime_get(), kstart));
+
        EXIT;
 out_child_unlock:
        if (object_locked)
@@ -1667,7 +1680,7 @@ static struct mdt_object *mdt_orphan_open(struct mdt_thread_info *info,
 
        uc = lu_ucred(env);
        uc_cap_save = uc->uc_cap;
-       uc->uc_cap |= 1 << CFS_CAP_DAC_OVERRIDE;
+       uc->uc_cap |= BIT(CFS_CAP_DAC_OVERRIDE);
        rc = mdo_create(env, mdt_object_child(local_root), &lname,
                        mdt_object_child(obj), spec, attr);
        uc->uc_cap = uc_cap_save;
@@ -2440,10 +2453,10 @@ int mdt_close(struct tgt_session_info *tsi)
        struct ptlrpc_request   *req = tgt_ses_req(tsi);
         struct md_attr         *ma = &info->mti_attr;
         struct mdt_body        *repbody = NULL;
+       ktime_t                 kstart = ktime_get();
         int rc, ret = 0;
         ENTRY;
 
-       mdt_counter_incr(req, LPROC_MDT_CLOSE);
        /* Close may come with the Size-on-MDS update. Unpack it. */
        rc = mdt_close_unpack(info);
        if (rc)
@@ -2498,5 +2511,8 @@ int mdt_close(struct tgt_session_info *tsi)
                tsi->tsi_reply_fail_id = OBD_FAIL_MDS_CLOSE_NET_REP;
 out:
        mdt_thread_info_fini(info);
+       if (rc == 0)
+               mdt_counter_incr(req, LPROC_MDT_CLOSE,
+                                ktime_us_delta(ktime_get(), kstart));
        RETURN(rc ? rc : ret);
 }