LU-15000 llog: read canceled records in llog_backup

llog_backup() do not reproduce index "holes" in the generated copy.
This could result to a llog copy indexes different from the source.
Then it might confuse the configuration update mechanism that rely on
indexes between the MGS source and the target copy.

This index gaps can be caused by "lctl --device MGS llog_cancel".

This patch add "raw" read mode to llog_process* to read canceled
records. So now llog_backup is able to reproduce an exact copy of
the original.

Signed-off-by: Etienne AUJAMES <etienne.aujames@cea.fr>
Change-Id: I811e23de8f4545bed36a44fedc2638d7418830dd
Reviewed-on: https://review.whamcloud.com/46552
Tested-by: jenkins <devops@whamcloud.com>
Reviewed-by: Dominique Martinet <qhufhnrynczannqp.f@noclue.notk.org>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: DELBARY Gael <gael.delbary@cea.fr>
Reviewed-by: Stephane Thiell <sthiell@stanford.edu>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h
index bde44be..360ba26 100644 (file)
--- a/lustre/include/lustre_log.h
+++ b/lustre/include/lustre_log.h
@@ -113,6 +113,13 @@ __u64 llog_size(const struct lu_env *env, struct llog_handle *llh);
 /* llog_process flags */
 #define LLOG_FLAG_NODEAMON 0x0001
 
+/* llog read mode, LLOG_READ_MODE_RAW will process llog canceled records */
+enum llog_read_mode {
+       LLOG_READ_MODE_NORMAL   = 0x0000,
+       LLOG_READ_MODE_RAW      = 0x0001,
+};
+
+
 /* llog_cat.c - catalog api */
 struct llog_process_data {
         /**
@@ -141,6 +148,10 @@ struct llog_process_cat_data {
          * Temporary stored last_idx while scanning log.
          */
         int                  lpcd_last_idx;
+       /**
+        * llog read mode
+        */
+       enum llog_read_mode  lpcd_read_mode;
 };
 
 int llog_cat_close(const struct lu_env *env, struct llog_handle *cathandle);

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index e021ee4..586ce9e 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -487,6 +487,15 @@ int llog_verify_record(const struct llog_handle *llh, struct llog_rec_hdr *rec)
 }
 EXPORT_SYMBOL(llog_verify_record);
 
+static inline bool llog_is_index_skipable(int idx, struct llog_log_hdr *llh,
+                                         struct llog_process_cat_data *cd)
+{
+       if (cd && (cd->lpcd_read_mode & LLOG_READ_MODE_RAW))
+               return false;
+
+       return !test_bit_le(idx, LLOG_HDR_BITMAP(llh));
+}
+
 static int llog_process_thread(void *arg)
 {
        struct llog_process_info        *lpi = arg;
@@ -524,8 +533,10 @@ static int llog_process_thread(void *arg)
                last_called_index = cd->lpcd_first_idx;
                index = cd->lpcd_first_idx + 1;
        }
-       if (cd != NULL && cd->lpcd_last_idx)
+       if (cd && cd->lpcd_last_idx)
                last_index = cd->lpcd_last_idx;
+       else if (cd && (cd->lpcd_read_mode & LLOG_READ_MODE_RAW))
+               last_index = loghandle->lgh_last_idx;
        else
                last_index = LLOG_HDR_BITMAP_SIZE(llh) - 1;
 
@@ -538,7 +549,7 @@ static int llog_process_thread(void *arg)
 
                /* skip records not set in bitmap */
                while (index <= last_index &&
-                      !test_bit_le(index, LLOG_HDR_BITMAP(llh)))
+                      llog_is_index_skipable(index, llh, cd))
                        ++index;
 
                /* There are no indices prior the last_index */
@@ -690,8 +701,8 @@ repeat:
                        loghandle->lgh_cur_offset = (char *)rec - (char *)buf +
                                                    chunk_offset;
 
-                       /* if set, process the callback on this record */
-                       if (test_bit_le(index, LLOG_HDR_BITMAP(llh))) {
+                       /* if needed, process the callback on this record */
+                       if (!llog_is_index_skipable(index, llh, cd)) {
                                struct llog_cookie *lgc;
                                __u64   tmp_off;
                                int     tmp_idx;
@@ -860,11 +871,12 @@ int llog_process_or_fork(const struct lu_env *env,
        lpi->lpi_cbdata    = data;
        lpi->lpi_catdata   = catdata;
 
-       CDEBUG(D_OTHER, "Processing "DFID" flags 0x%03x startcat %d startidx %d first_idx %d last_idx %d\n",
+       CDEBUG(D_OTHER, "Processing "DFID" flags 0x%03x startcat %d startidx %d first_idx %d last_idx %d read_mode %d\n",
               PFID(&loghandle->lgh_id.lgl_oi.oi_fid), flags,
               (flags & LLOG_F_IS_CAT) && d ? d->lpd_startcat : -1,
               (flags & LLOG_F_IS_CAT) && d ? d->lpd_startidx : -1,
-              cd ? cd->lpcd_first_idx : -1, cd ? cd->lpcd_last_idx : -1);
+              cd ? cd->lpcd_first_idx : -1, cd ? cd->lpcd_last_idx : -1,
+              cd ? cd->lpcd_read_mode : -1);
        if (fork) {
                struct task_struct *task;
 
@@ -955,7 +967,7 @@ int llog_reverse_process(const struct lu_env *env,
 
                /* skip records not set in bitmap */
                while (index >= first_index &&
-                      !test_bit_le(index, LLOG_HDR_BITMAP(llh)))
+                      llog_is_index_skipable(index, llh, cd))
                        --index;
 
                LASSERT(index >= first_index - 1);
@@ -985,8 +997,8 @@ int llog_reverse_process(const struct lu_env *env,
                        if (tail->lrt_index == 0)
                                GOTO(out, rc = 0); /* no more records */
 
-                       /* if set, process the callback on this record */
-                       if (test_bit_le(index, LLOG_HDR_BITMAP(llh))) {
+                       /* if needed, process the callback on this record */
+                       if (!llog_is_index_skipable(index, llh, cd)) {
                                rec = (void *)tail - tail->lrt_len +
                                      sizeof(*tail);
 
@@ -1417,13 +1429,24 @@ out:
 }
 EXPORT_SYMBOL(llog_is_empty);
 
+/* this callback run in raw read mode (canceled record are processed) */
 int llog_copy_handler(const struct lu_env *env, struct llog_handle *llh,
                      struct llog_rec_hdr *rec, void *data)
 {
-       struct llog_handle      *copy_llh = data;
+       struct llog_handle *copy_llh = data;
+       int idx = rec->lrh_index;
+       int rc;
+
+       ENTRY;
 
        /* Append all records */
-       return llog_write(env, copy_llh, rec, LLOG_NEXT_IDX);
+       rc = llog_write(env, copy_llh, rec, LLOG_NEXT_IDX);
+
+       /* Cancel the record if it is canceled on the source */
+       if (!rc && !test_bit_le(idx, LLOG_HDR_BITMAP(llh->lgh_hdr)))
+               rc = llog_cancel_rec(env, copy_llh, copy_llh->lgh_last_idx);
+
+       RETURN(rc);
 }
 
 /* backup plain llog */
@@ -1431,8 +1454,9 @@ int llog_backup(const struct lu_env *env, struct obd_device *obd,
                struct llog_ctxt *ctxt, struct llog_ctxt *bctxt,
                char *name, char *backup)
 {
-       struct llog_handle      *llh, *bllh;
-       int                      rc;
+       struct llog_handle *llh, *bllh;
+       struct llog_process_cat_data cd = {0};
+       int rc;
 
        ENTRY;
 
@@ -1477,9 +1501,11 @@ int llog_backup(const struct lu_env *env, struct obd_device *obd,
        if (rc)
                GOTO(out_backup, rc);
 
+       /* Read canceled records to have an exact copy */
+       cd.lpcd_read_mode = LLOG_READ_MODE_RAW;
        /* Copy log record by record */
        rc = llog_process_or_fork(env, llh, llog_copy_handler, (void *)bllh,
-                                 NULL, false);
+                                 &cd, false);
        if (rc)
                CERROR("%s: failed to backup log %s: rc = %d\n",
                       obd->obd_name, name, rc);

diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c
index 3f53dab..ba44ad3 100644 (file)
--- a/lustre/obdclass/llog_cat.c
+++ b/lustre/obdclass/llog_cat.c
@@ -868,6 +868,7 @@ static int llog_cat_process_cb(const struct lu_env *env,
        } else if (d->lpd_startidx > 0) {
                 struct llog_process_cat_data cd;
 
+                cd.lpcd_read_mode = LLOG_READ_MODE_NORMAL;
                 cd.lpcd_first_idx = d->lpd_startidx;
                 cd.lpcd_last_idx = 0;
                rc = llog_process_or_fork(env, llh, d->lpd_cb, d->lpd_data,
@@ -920,7 +921,9 @@ int llog_cat_process_or_fork(const struct lu_env *env,
 
        if (llh->llh_cat_idx >= cat_llh->lgh_last_idx &&
            llh->llh_count > 1) {
-               struct llog_process_cat_data cd;
+               struct llog_process_cat_data cd = {
+                       .lpcd_read_mode = LLOG_READ_MODE_NORMAL
+               };
 
                CWARN("%s: catlog "DFID" crosses index zero\n",
                      loghandle2name(cat_llh),
@@ -1083,6 +1086,7 @@ int llog_cat_reverse_process(const struct lu_env *env,
         ENTRY;
 
         LASSERT(llh->llh_flags & LLOG_F_IS_CAT);
+       cd.lpcd_read_mode = LLOG_READ_MODE_NORMAL;
         d.lpd_data = data;
         d.lpd_cb = cb;
 

diff --git a/lustre/obdclass/llog_test.c b/lustre/obdclass/llog_test.c
index ef0d146..e6d3634 100644 (file)
--- a/lustre/obdclass/llog_test.c
+++ b/lustre/obdclass/llog_test.c
@@ -380,6 +380,7 @@ static int llog_test3_process(const struct lu_env *env,
 
        CWARN("test3: processing records from index %d to the end\n",
              start);
+       cd.lpcd_read_mode = LLOG_READ_MODE_NORMAL;
        cd.lpcd_first_idx = start - 1;
        cd.lpcd_last_idx = 0;
        test_3_rec_num = test_3_paddings = 0;

diff --git a/lustre/obdclass/obd_config.c b/lustre/obdclass/obd_config.c
index c450c07..07e482e 100644 (file)
--- a/lustre/obdclass/obd_config.c
+++ b/lustre/obdclass/obd_config.c
@@ -2010,6 +2010,7 @@ int class_config_parse_llog(const struct lu_env *env, struct llog_ctxt *ctxt,
 {
        struct llog_process_cat_data cd = {
                .lpcd_first_idx = 0,
+               .lpcd_read_mode = LLOG_READ_MODE_NORMAL,
        };
        struct llog_handle *llh;
        llog_cb_t callback;