From: Niu Yawei Date: Mon, 24 Sep 2012 08:34:02 +0000 (-0400) Subject: LU-1842 quota: migrate old quota admin files X-Git-Tag: 2.3.51~36 X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=commitdiff_plain;h=3ff4a0744c229e0199bc7d93db9221c3bfb1f846 LU-1842 quota: migrate old quota admin files Migrate old admin quota files into new global indexes when mount with an old device. Signed-off-by: Niu Yawei Change-Id: I9e834eabf83f0da273b4ba3309adf71f55eae758 Reviewed-on: http://review.whamcloud.com/4077 Tested-by: Hudson Reviewed-by: Johann Lombardi Tested-by: Maloo Reviewed-by: Andreas Dilger --- diff --git a/lustre/osd-ldiskfs/osd_handler.c b/lustre/osd-ldiskfs/osd_handler.c index 32ca46e..b5ab966 100644 --- a/lustre/osd-ldiskfs/osd_handler.c +++ b/lustre/osd-ldiskfs/osd_handler.c @@ -2828,6 +2828,9 @@ static int osd_index_try(const struct lu_env *env, struct dt_object *dt, } LINVRNT(osd_invariant(obj)); + if (is_quota_glb_feat(feat)) + result = osd_quota_migration(env, dt, feat); + return result; } diff --git a/lustre/osd-ldiskfs/osd_internal.h b/lustre/osd-ldiskfs/osd_internal.h index 9efb06f..ab5fc62a 100644 --- a/lustre/osd-ldiskfs/osd_internal.h +++ b/lustre/osd-ldiskfs/osd_internal.h @@ -694,6 +694,16 @@ const struct dt_rec *osd_quota_pack(struct osd_object *obj, const struct dt_rec *rec, union lquota_rec *quota_rec); void osd_quota_unpack(struct osd_object *obj, const struct dt_rec *rec); +int osd_quota_migration(const struct lu_env *env, struct dt_object *dt, + const struct dt_index_features *feat); + +static inline bool is_quota_glb_feat(const struct dt_index_features *feat) +{ + return (feat == &dt_quota_iusr_features || + feat == &dt_quota_busr_features || + feat == &dt_quota_igrp_features || + feat == &dt_quota_bgrp_features) ? true : false; +} /* * Invariants, assertions. diff --git a/lustre/osd-ldiskfs/osd_quota.c b/lustre/osd-ldiskfs/osd_quota.c index 26257ea..8dbe02b 100644 --- a/lustre/osd-ldiskfs/osd_quota.c +++ b/lustre/osd-ldiskfs/osd_quota.c @@ -584,3 +584,543 @@ int osd_declare_inode_qid(const struct lu_env *env, qid_t uid, qid_t gid, RETURN(rcu ? rcu : rcg); } + +/* Following code is used to migrate old admin quota files (in Linux quota + * file v2 format) into the new quota global indexes (in IAM format). */ + +#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2,7,50,0) + +/* copied from osd_it_acct_get(), only changed the 'type' to -1 */ +static int osd_it_admin_get(const struct lu_env *env, struct dt_it *di, + const struct dt_key *key) +{ + struct osd_it_quota *it = (struct osd_it_quota *)di; + int type = -1; + qid_t dqid = *(qid_t *)key; + loff_t offset; + int rc; + ENTRY; + + offset = find_tree_dqentry(env, it->oiq_obj, type, dqid, + LUSTRE_DQTREEOFF, 0, it); + if (offset > 0) { /* Found */ + RETURN(+1); + } else if (offset < 0) { /* Error */ + QUOTA_IT_READ_ERROR(it, (int)offset); + RETURN((int)offset); + } + + /* The @key is not found, move to the first valid entry */ + rc = walk_tree_dqentry(env, it->oiq_obj, type, it->oiq_blk[0], 0, + 0, it); + if (rc > 0) + /* no valid entry found */ + rc = -ENOENT; + RETURN(rc); +} + +static int osd_it_admin_load(const struct lu_env *env, + const struct dt_it *di, __u64 hash) +{ + int rc; + ENTRY; + + rc = osd_it_admin_get(env, (struct dt_it *)di, + (const struct dt_key *)&hash); + RETURN(rc); +} + +static int osd_it_admin_rec(const struct lu_env *env, + const struct dt_it *di, + struct dt_rec *dtrec, __u32 attr) +{ + struct osd_it_quota *it = (struct osd_it_quota *)di; + struct lu_buf buf; + loff_t pos; + int rc; + struct lustre_disk_dqblk_v2 *dqblk = + (struct lustre_disk_dqblk_v2 *)dtrec; + ENTRY; + + buf.lb_buf = dqblk; + buf.lb_len = sizeof(*dqblk); + + pos = it->oiq_offset; + rc = dt_record_read(env, &it->oiq_obj->oo_dt, &buf, &pos); + RETURN(rc); +} + +/* copied from osd_it_acct_next(), only changed the 'type' to -1 */ +static int osd_it_admin_next(const struct lu_env *env, struct dt_it *di) +{ + struct osd_it_quota *it = (struct osd_it_quota *)di; + int type = -1; + int depth, rc; + uint index; + ENTRY; + + /* Let's first check if there are any remaining valid entry in the + * current leaf block. Start with the next entry after the current one. + */ + depth = LUSTRE_DQTREEDEPTH; + index = it->oiq_index[depth]; + if (++index < LUSTRE_DQSTRINBLK) { + /* Search for the next valid entry from current index */ + rc = walk_block_dqentry(env, it->oiq_obj, type, + it->oiq_blk[depth], index, it); + if (rc < 0) { + QUOTA_IT_READ_ERROR(it, rc); + RETURN(rc); + } else if (rc == 0) { + /* Found on entry, @it is already updated to the + * new position in walk_block_dqentry(). */ + RETURN(0); + } else { + rc = osd_it_add_processed(it, depth); + if (rc) + RETURN(rc); + } + } else { + rc = osd_it_add_processed(it, depth); + if (rc) + RETURN(rc); + } + rc = 1; + + /* We have consumed all the entries of the current leaf block, move on + * to the next one. */ + depth--; + + /* We keep searching as long as walk_tree_dqentry() returns +1 + * (= no valid entry found). */ + for (; depth >= 0 && rc > 0; depth--) { + index = it->oiq_index[depth]; + if (++index > 0xff) + continue; + rc = walk_tree_dqentry(env, it->oiq_obj, type, + it->oiq_blk[depth], depth, index, it); + } + + if (rc < 0) + QUOTA_IT_READ_ERROR(it, rc); + RETURN(rc); +} + +const struct dt_index_operations osd_admin_index_ops = { + .dio_lookup = osd_acct_index_lookup, + .dio_it = { + .init = osd_it_acct_init, + .fini = osd_it_acct_fini, + .get = osd_it_admin_get, + .put = osd_it_acct_put, + .next = osd_it_admin_next, + .key = osd_it_acct_key, + .key_size = osd_it_acct_key_size, + .rec = osd_it_admin_rec, + .store = osd_it_acct_store, + .load = osd_it_admin_load + } +}; + +static int write_quota_rec(const struct lu_env *env, struct dt_object *dt, + __u64 id, struct lquota_glb_rec *rec) +{ + struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt)); + struct thandle *th; + struct dt_key *key = (struct dt_key *)&id; + int rc; + ENTRY; + + th = dt_trans_create(env, &osd->od_dt_dev); + if (IS_ERR(th)) + RETURN(PTR_ERR(th)); + + /* the entry with 0 key can always be found in IAM file. */ + if (id == 0) { + rc = dt_declare_delete(env, dt, key, th); + if (rc) + GOTO(out, rc); + } + + rc = dt_declare_insert(env, dt, (struct dt_rec *)rec, key, th); + if (rc) + GOTO(out, rc); + + rc = dt_trans_start_local(env, &osd->od_dt_dev, th); + if (rc) + GOTO(out, rc); + + dt_write_lock(env, dt, 0); + + if (id == 0) { + struct lquota_glb_rec *tmp; + + OBD_ALLOC_PTR(tmp); + if (tmp == NULL) + GOTO(out_lock, rc = -ENOMEM); + + rc = dt_lookup(env, dt, (struct dt_rec *)tmp, key, + BYPASS_CAPA); + + OBD_FREE_PTR(tmp); + if (rc == 0) { + rc = dt_delete(env, dt, key, th, BYPASS_CAPA); + if (rc) + GOTO(out_lock, rc); + } + rc = 0; + } + + rc = dt_insert(env, dt, (struct dt_rec *)rec, key, th, BYPASS_CAPA, 1); +out_lock: + dt_write_unlock(env, dt); +out: + dt_trans_stop(env, &osd->od_dt_dev, th); + RETURN(rc); +} + +static int convert_quota_file(const struct lu_env *env, + struct dt_object *old, struct dt_object *new, + bool isblk) +{ + const struct dt_it_ops *iops = &old->do_index_ops->dio_it; + struct osd_object *obj; + struct lu_buf buf; + struct dt_it *it; + struct dt_key *key; + __u32 grace; + struct lquota_glb_rec *glb_rec = NULL; + loff_t pos; + int rc; + struct lustre_disk_dqblk_v2 *dqblk = NULL; + struct lustre_disk_dqinfo *dqinfo = NULL; + ENTRY; + + obj = osd_dt_obj(old); + LASSERT(obj->oo_inode); + + if (i_size_read(obj->oo_inode) == 0) + RETURN(0); + + /* allocate buffers */ + OBD_ALLOC_PTR(dqinfo); + if (dqinfo == NULL) + RETURN(-ENOMEM); + + OBD_ALLOC_PTR(glb_rec); + if (glb_rec == NULL) + GOTO(out, rc = -ENOMEM); + + OBD_ALLOC_PTR(dqblk); + if (dqblk == NULL) + GOTO(out, rc = -ENOMEM); + + /* convert the old igrace/bgrace */ + buf.lb_buf = dqinfo; + buf.lb_len = sizeof(*dqinfo); + pos = LUSTRE_DQINFOOFF; + + rc = dt_record_read(env, old, &buf, &pos); + if (rc) + GOTO(out, rc); + + /* keep it in little endian */ + grace = isblk ? dqinfo->dqi_bgrace : dqinfo->dqi_igrace; + if (grace != 0) { + glb_rec->qbr_time = grace; + rc = write_quota_rec(env, new, 0, glb_rec); + if (rc) + GOTO(out, rc); + glb_rec->qbr_time = 0; + } + + /* iterate the old admin file, insert each record into the + * new index file. */ + it = iops->init(env, old, 0, BYPASS_CAPA); + if (IS_ERR(it)) + GOTO(out, rc = PTR_ERR(it)); + + rc = iops->load(env, it, 0); + if (rc == -ENOENT) + GOTO(out_it, rc = 0); + else if (rc < 0) + GOTO(out_it, rc); + + do { + key = iops->key(env, it); + if (IS_ERR(key)) + GOTO(out_it, rc = PTR_ERR(key)); + + /* skip the root user/group */ + if (*((__u64 *)key) == 0) + goto next; + + rc = iops->rec(env, it, (struct dt_rec *)dqblk, 0); + if (rc) + GOTO(out_it, rc); + + /* keep the value in little endian */ + glb_rec->qbr_hardlimit = isblk ? dqblk->dqb_bhardlimit : + dqblk->dqb_ihardlimit; + glb_rec->qbr_softlimit = isblk ? dqblk->dqb_bsoftlimit : + dqblk->dqb_isoftlimit; + + rc = write_quota_rec(env, new, *((__u64 *)key), glb_rec); + if (rc) + GOTO(out_it, rc); +next: + rc = iops->next(env, it); + } while (rc == 0); + + /* reach the end */ + if (rc > 0) + rc = 0; + +out_it: + iops->put(env, it); + iops->fini(env, it); +out: + if (dqblk != NULL) + OBD_FREE_PTR(dqblk); + if (glb_rec != NULL) + OBD_FREE_PTR(glb_rec); + if (dqinfo != NULL) + OBD_FREE_PTR(dqinfo); + return rc; +} + +static int truncate_quota_index(const struct lu_env *env, struct dt_object *dt, + const struct dt_index_features *feat) +{ + struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt)); + struct thandle *th; + struct lu_attr *attr; + struct osd_thandle *oth; + struct inode *inode; + int rc; + ENTRY; + + OBD_ALLOC_PTR(attr); + if (attr == NULL) + RETURN(-ENOMEM); + + attr->la_size = 0; + attr->la_valid = LA_SIZE; + + th = dt_trans_create(env, &osd->od_dt_dev); + if (IS_ERR(th)) { + OBD_FREE_PTR(attr); + RETURN(PTR_ERR(th)); + } + + rc = dt_declare_punch(env, dt, 0, OBD_OBJECT_EOF, th); + if (rc) + GOTO(out, rc); + + rc = dt_declare_attr_set(env, dt, attr, th); + if (rc) + GOTO(out, rc); + + inode = osd_dt_obj(dt)->oo_inode; + LASSERT(inode); + + rc = dt_declare_record_write(env, dt, inode->i_sb->s_blocksize * 2, 0, th); + if (rc) + GOTO(out, rc); + + rc = dt_trans_start_local(env, &osd->od_dt_dev, th); + if (rc) + GOTO(out, rc); + + dt_write_lock(env, dt, 0); + rc = dt_punch(env, dt, 0, OBD_OBJECT_EOF, th, BYPASS_CAPA); + if (rc) + GOTO(out_lock, rc); + + rc = dt_attr_set(env, dt, attr, th, BYPASS_CAPA); + if (rc) + GOTO(out_lock, rc); + + oth = container_of(th, struct osd_thandle, ot_super); + + if (feat->dif_flags & DT_IND_VARKEY) + rc = iam_lvar_create(osd_dt_obj(dt)->oo_inode, + feat->dif_keysize_max, + feat->dif_ptrsize, + feat->dif_recsize_max, oth->ot_handle); + else + rc = iam_lfix_create(osd_dt_obj(dt)->oo_inode, + feat->dif_keysize_max, + feat->dif_ptrsize, + feat->dif_recsize_max, oth->ot_handle); +out_lock: + dt_write_unlock(env, dt); +out: + dt_trans_stop(env, &osd->od_dt_dev, th); + OBD_FREE_PTR(attr); + RETURN(rc); +} + +static int set_quota_index_version(const struct lu_env *env, + struct dt_object *dt) +{ + struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt)); + struct thandle *th; + int rc; + ENTRY; + + th = dt_trans_create(env, &osd->od_dt_dev); + if (IS_ERR(th)) + RETURN(PTR_ERR(th)); + + rc = dt_declare_version_set(env, dt, th); + if (rc) + GOTO(out, rc); + + rc = dt_trans_start_local(env, &osd->od_dt_dev, th); + if (rc) + GOTO(out, rc); + + th->th_sync = 1; + dt_version_set(env, dt, 1, th); +out: + dt_trans_stop(env, &osd->od_dt_dev, th); + RETURN(rc); +} + +#define OBJECTS "OBJECTS" +#define ADMIN_USR "admin_quotafile_v2.usr" +#define ADMIN_GRP "admin_quotafile_v2.grp" + +int osd_quota_migration(const struct lu_env *env, struct dt_object *dt, + const struct dt_index_features *feat) +{ + struct osd_thread_info *oti = osd_oti_get(env); + struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt)); + struct dt_object *root, *parent = NULL, *admin = NULL; + dt_obj_version_t version; + char *fname; + bool isblk; + int rc; + ENTRY; + + /* not newly created global index */ + version = dt_version_get(env, dt); + if (version != 0) + RETURN(0); + + /* locate root */ + rc = dt_root_get(env, &osd->od_dt_dev, &oti->oti_fid); + if (rc) { + CERROR("%s: Can't get root FID, rc:%d\n", osd->od_svname, rc); + RETURN(rc); + } + + root = dt_locate(env, &osd->od_dt_dev, &oti->oti_fid); + if (IS_ERR(root)) { + CERROR("%s: Failed to locate root "DFID", rc:%ld\n", + osd->od_svname, PFID(&oti->oti_fid), PTR_ERR(root)); + RETURN(PTR_ERR(root)); + } + + /* locate /OBJECTS */ + rc = dt_lookup_dir(env, root, OBJECTS, &oti->oti_fid); + if (rc == -ENOENT) { + GOTO(out, rc = 0); + } else if (rc) { + CERROR("%s: Failed to lookup %s, rc:%d\n", + osd->od_svname, OBJECTS, rc); + GOTO(out, rc); + } + + parent = dt_locate(env, &osd->od_dt_dev, &oti->oti_fid); + if (IS_ERR(parent)) { + CERROR("%s: Failed to locate %s "DFID", rc:%ld\n", + osd->od_svname, OBJECTS, PFID(&oti->oti_fid), + PTR_ERR(parent)); + GOTO(out, rc = PTR_ERR(parent)); + } + + /* locate quota admin file */ + if (feat == &dt_quota_iusr_features) { + fname = ADMIN_USR; + isblk = false; + } else if (feat == &dt_quota_busr_features) { + fname = ADMIN_USR; + isblk = true; + } else if (feat == &dt_quota_igrp_features) { + fname = ADMIN_GRP; + isblk = false; + } else { + fname = ADMIN_GRP; + isblk = true; + } + + rc = dt_lookup_dir(env, parent, fname, &oti->oti_fid); + if (rc == -ENOENT) { + GOTO(out, rc = 0); + } else if (rc) { + CERROR("%s: Failed to lookup %s, rc:%d\n", + osd->od_svname, fname, rc); + GOTO(out, rc); + } + + admin = dt_locate(env, &osd->od_dt_dev, &oti->oti_fid); + if (IS_ERR(admin)) { + CERROR("%s: Failed to locate %s "DFID", rc:%d\n", + osd->od_svname, fname, PFID(&oti->oti_fid), rc); + GOTO(out, rc = PTR_ERR(admin)); + } + + if (!dt_object_exists(admin)) { + CERROR("%s: Old admin file %s doesn't exist, but is still " + " referenced in parent directory.\n", + osd->od_svname, fname); + GOTO(out, rc = -ENOENT); + } + + /* truncate the new quota index file in case of any leftovers + * from last failed migration */ + rc = truncate_quota_index(env, dt, feat); + if (rc) { + CERROR("%s: Failed to truncate the quota index "DFID", rc:%d\n", + osd->od_svname, PFID(lu_object_fid(&dt->do_lu)), rc); + RETURN(rc); + } + + /* set up indexing operations for the admin file */ + admin->do_index_ops = &osd_admin_index_ops; + + LCONSOLE_INFO("%s: Migrate %s quota from old admin quota file(%s) to " + "new IAM quota index("DFID").\n", osd->od_svname, + isblk ? "block" : "inode", fname, + PFID(lu_object_fid(&dt->do_lu))); + + /* iterate the admin quota file, and insert each record into + * the new index file */ + rc = convert_quota_file(env, admin, dt, isblk); + if (rc) + CERROR("%s: Migrate old admin quota file(%s) failed, rc:%d\n", + osd->od_svname, fname, rc); +out: + /* bump index version to 1, so the migration will be skipped + * next time. */ + if (rc == 0) { + rc = set_quota_index_version(env , dt); + if (rc) + CERROR("%s: Failed to set quota index("DFID") " + "version, rc:%d\n", osd->od_svname, + PFID(lu_object_fid(&dt->do_lu)), rc); + } + + if (admin && !IS_ERR(admin)) + lu_object_put(env, &admin->do_lu); + if (parent && !IS_ERR(parent)) + lu_object_put(env, &parent->do_lu); + lu_object_put(env, &root->do_lu); + + RETURN(rc); +} +#else +#warning "remove old quota compatibility code" +#endif diff --git a/lustre/osd-ldiskfs/osd_quota_fmt.c b/lustre/osd-ldiskfs/osd_quota_fmt.c index 25db1ce..45f3e08 100644 --- a/lustre/osd-ldiskfs/osd_quota_fmt.c +++ b/lustre/osd-ldiskfs/osd_quota_fmt.c @@ -64,6 +64,29 @@ static ssize_t quota_read_blk(const struct lu_env *env, ENTRY; memset(buf, 0, LUSTRE_DQBLKSIZE); + +#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2,7,50,0) + /* type is set as -1 when reading old admin quota file */ + if (type != USRQUOTA && type != GRPQUOTA) { + struct lu_buf lu_buffer; + loff_t pos; + + lu_buffer.lb_buf = buf; + lu_buffer.lb_len = LUSTRE_DQBLKSIZE; + pos = blk << LUSTRE_DQBLKSIZE_BITS; + + ret = dt_record_read(env, &obj->oo_dt, &lu_buffer, &pos); + + if (ret == 0) + ret = LUSTRE_DQBLKSIZE; + else if (ret == -EBADR || ret == -EFAULT) + ret = 0; + RETURN(ret); + } +#else +#warning "remove old quota compatibility code" +#endif + ret = sb->s_op->quota_read(sb, type, buf, LUSTRE_DQBLKSIZE, blk << LUSTRE_DQBLKSIZE_BITS);