LU-8320 llog: prevent llog ID re-use.

The new llog ID generated in simple manner by
increment of the last used OID within the same FID
sequence. Therefore OID can cycle through zero and
match some very old llog file which is still in use
or orphaned. Patch skips such llog IDs. Also patch
replaces several assertions with sanity checks in
llog_write() path.

Signed-off-by: Mikhal Pershin <mike.pershin@intel.com>
Change-Id: Iffd7e61f75fdb554ae03c8a4b685cd2aff648c24
Reviewed-on: http://review.whamcloud.com/21144
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Tested-by: Jenkins
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index 0ab7a34..f7c398e 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -898,11 +898,29 @@ int llog_write_rec(const struct lu_env *env, struct llog_handle *handle,
 
        ENTRY;
 
+       /* API sanity checks */
+       if (handle == NULL) {
+               CERROR("loghandle is missed\n");
+               RETURN(-EPROTO);
+       } else if (handle->lgh_obj == NULL) {
+               CERROR("loghandle %p with NULL object\n",
+                       handle);
+               RETURN(-EPROTO);
+       } else if (th == NULL) {
+               CERROR("%s: missed transaction handle\n",
+                       handle->lgh_obj->do_lu.lo_dev->ld_obd->obd_name);
+               RETURN(-EPROTO);
+       } else if (handle->lgh_hdr == NULL) {
+               CERROR("%s: loghandle %p with no header\n",
+                       handle->lgh_obj->do_lu.lo_dev->ld_obd->obd_name,
+                       handle);
+               RETURN(-EPROTO);
+       }
+
        rc = llog_handle2ops(handle, &lop);
        if (rc)
                RETURN(rc);
 
-       LASSERT(lop);
        if (lop->lop_write_rec == NULL)
                RETURN(-EOPNOTSUPP);
 

diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c
index 8432d39..6027ef2 100644 (file)
--- a/lustre/obdclass/llog_osd.c
+++ b/lustre/obdclass/llog_osd.c
@@ -387,12 +387,8 @@ static int llog_osd_write_rec(const struct lu_env *env,
        __u64                   orig_write_offset;
        ENTRY;
 
-       LASSERT(env);
        llh = loghandle->lgh_hdr;
-       LASSERT(llh);
        o = loghandle->lgh_obj;
-       LASSERT(o);
-       LASSERT(th);
 
        chunk_size = llh->llh_hdr.lrh_len;
        CDEBUG(D_OTHER, "new record %x to "DFID"\n",
@@ -1176,6 +1172,7 @@ static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
        struct ls_device                *ls;
        struct local_oid_storage        *los = NULL;
        int                              rc = 0;
+       bool new_id = false;
 
        ENTRY;
 
@@ -1236,6 +1233,7 @@ static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
                        /* generate fid for new llog */
                        rc = local_object_fid_generate(env, los,
                                                       &lgi->lgi_fid);
+                       new_id = true;
                }
                if (rc < 0)
                        GOTO(out, rc);
@@ -1247,15 +1245,30 @@ static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
        } else {
                LASSERTF(open_param & LLOG_OPEN_NEW, "%#x\n", open_param);
                /* generate fid for new llog */
+generate:
                rc = local_object_fid_generate(env, los, &lgi->lgi_fid);
                if (rc < 0)
                        GOTO(out, rc);
+               new_id = true;
        }
 
        o = ls_locate(env, ls, &lgi->lgi_fid, NULL);
        if (IS_ERR(o))
                GOTO(out_name, rc = PTR_ERR(o));
 
+       if (dt_object_exists(o) && new_id) {
+               /* llog exists with just generated ID, e.g. some old llog file
+                * still is in use or is orphan, drop a warn and skip it. */
+               CDEBUG(D_INFO, "%s: llog exists with the same FID: "DFID
+                      ", skipping\n",
+                      o->do_lu.lo_dev->ld_obd->obd_name,
+                      PFID(lu_object_fid(&o->do_lu)));
+               lu_object_put(env, &o->do_lu);
+               /* just skip this llog ID, we shouldn't delete it because we
+                * don't know exactly what is its purpose and state. */
+               goto generate;
+       }
+
 after_open:
        /* No new llog is expected but doesn't exist */
        if (open_param != LLOG_OPEN_NEW && !dt_object_exists(o))