LU-13974 llog: check stale osp object

[fs/lustre-release.git] / lustre / obdclass / llog_osd.c

diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c
index edc1edc..92d7186 100644 (file)
--- a/lustre/obdclass/llog_osd.c
+++ b/lustre/obdclass/llog_osd.c
@@ -412,6 +412,9 @@ static int llog_osd_write_rec(const struct lu_env *env,
                LASSERT(llh->llh_size == reclen);
        }
 
+       /* return error if osp object is stale */
+       if (idx != LLOG_HEADER_IDX && dt_object_stale(o))
+               RETURN(-ESTALE);
        rc = dt_attr_get(env, o, &lgi->lgi_attr);
        if (rc)
                RETURN(rc);
@@ -612,8 +615,9 @@ static int llog_osd_write_rec(const struct lu_env *env,
         * update/cancel, the llh_count and llh_bitmap are protected */
        mutex_lock(&loghandle->lgh_hdr_mutex);
        if (__test_and_set_bit_le(index, LLOG_HDR_BITMAP(llh))) {
-               CERROR("%s: index %u already set in log bitmap\n",
-                      o->do_lu.lo_dev->ld_obd->obd_name, index);
+               CERROR("%s: index %u already set in llog bitmap "DFID"\n",
+                      o->do_lu.lo_dev->ld_obd->obd_name, index,
+                      PFID(lu_object_fid(&o->do_lu)));
                mutex_unlock(&loghandle->lgh_hdr_mutex);
                LBUG(); /* should never happen */
        }
@@ -967,9 +971,25 @@ static int llog_osd_next_block(const struct lu_env *env,
                rec = buf;
                if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
                        lustre_swab_llog_rec(rec);
-
                tail = (struct llog_rec_tail *)((char *)buf + rc -
                                                sizeof(struct llog_rec_tail));
+
+               if (llog_verify_record(loghandle, rec)) {
+                       /*
+                        * the block seems corrupted. make a pad record so the
+                        * caller can skip the block and try with the next one
+                        */
+                       rec->lrh_len = rc;
+                       rec->lrh_index = next_idx;
+                       rec->lrh_type = LLOG_PAD_MAGIC;
+
+                       tail = rec_tail(rec);
+                       tail->lrt_len = rc;
+                       tail->lrt_index = next_idx;
+
+                       GOTO(out, rc = 0);
+               }
+
                /* get the last record in block */
                last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
                                                   tail->lrt_len);
@@ -1006,7 +1026,7 @@ static int llog_osd_next_block(const struct lu_env *env,
 
                /* sanity check that the start of the new buffer is no farther
                 * than the record that we wanted.  This shouldn't happen. */
-               if (rec->lrh_index > next_idx) {
+               if (next_idx && rec->lrh_index > next_idx) {
                        if (!force_mini_rec && next_idx > last_idx)
                                goto retry;
 
@@ -1247,8 +1267,7 @@ static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
                } else {
                        /* If logid == NULL, then it means the caller needs
                         * to allocate new FID (llog_cat_declare_add_rec()). */
-                       rc = obd_fid_alloc(env, ctxt->loc_exp,
-                                          &lgi->lgi_fid, NULL);
+                       rc = dt_fid_alloc(env, dt, &lgi->lgi_fid, NULL, NULL);
                        if (rc < 0)
                                RETURN(rc);
                        rc = 0;
@@ -1309,7 +1328,12 @@ generate:
                        GOTO(out, rc);
                new_id = true;
        }
-
+       if (OBD_FAIL_PRECHECK(OBD_FAIL_MDS_LLOG_UMOUNT_RACE) &&
+           cfs_fail_val == 1) {
+               cfs_fail_val = 2;
+               OBD_RACE(OBD_FAIL_MDS_LLOG_UMOUNT_RACE);
+               msleep(MSEC_PER_SEC);
+       }
        o = ls_locate(env, ls, &lgi->lgi_fid, NULL);
        if (IS_ERR(o))
                GOTO(out_name, rc = PTR_ERR(o));
@@ -2010,7 +2034,7 @@ int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
                if (IS_ERR(th))
                        GOTO(out, rc = PTR_ERR(th));
 
-               lgi->lgi_attr.la_valid = LA_MODE;
+               lgi->lgi_attr.la_valid = LA_MODE | LA_TYPE;
                lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
                lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);