LU-12275 sec: deal with encrypted object size

Problem with size of encrypted file comes from the fact that
an encrypted page will always contain PAGE_SIZE bytes of data,
even if clear text page is only a few bytes. And server infers
object size from content of encrypted page.

The way to address this is the following. Upon writing, when the
client encrypts the page representing the end of the file, it puts
into o_size info of the request's body, the size of the clear text
version of the file. On server side, this information is used to
adjust isize of the object, but still storing the complete pages
on disk.

Signed-off-by: Sebastien Buisson <sbuisson@ddn.com>
Change-Id: Ia83424123da26920ba0e0dfb354f54b1fa0ccfbb
Reviewed-on: https://review.whamcloud.com/36146
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: John L. Hammond <jhammond@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/dt_object.h b/lustre/include/dt_object.h
index d9cc933..c1c5d28 100644 (file)
--- a/lustre/include/dt_object.h
+++ b/lustre/include/dt_object.h
@@ -1317,11 +1317,21 @@ struct dt_body_operations {
         * then the method should maintain space accounting for the given
         * credentials.
         *
         * then the method should maintain space accounting for the given
         * credentials.
         *

+        * user_size parameter is the apparent size of the file, ie the size
+        * of the clear text version of the file. It can differ from the actual
+        * amount of valuable data received when a file is encrypted,
+        * because encrypted pages always contain PAGE_SIZE bytes of data,
+        * even if clear text data is only a few bytes.
+        * In case of encrypted file, apparent size will be stored as the inode
+        * size, so that servers return to clients an object size they can use
+        * to determine clear text size.
+        *

         * \param[in] env       execution environment for this thread
         * \param[in] dt        object
         * \param[in] lb        array of descriptors for the buffers
         * \param[in] nr        size of the array
         * \param[in] th        transaction handle
         * \param[in] env       execution environment for this thread
         * \param[in] dt        object
         * \param[in] lb        array of descriptors for the buffers
         * \param[in] nr        size of the array
         * \param[in] th        transaction handle

+        * \param[in] user_size apparent size

         *
         * \retval 0            on success
         * \retval negative     negated errno on error
         *
         * \retval 0            on success
         * \retval negative     negated errno on error


@@ -1330,7 +1340,8 @@ struct dt_body_operations {
                                struct dt_object *dt,
                                struct niobuf_local *lb,
                                int nr,
                                struct dt_object *dt,
                                struct niobuf_local *lb,
                                int nr,

-                               struct thandle *th);
+                               struct thandle *th,
+                               __u64 user_size);

 
        /**
         * Return logical to physical block mapping for a given extent
 
        /**
         * Return logical to physical block mapping for a given extent


@@ -2495,13 +2506,13 @@ static inline int dt_declare_write_commit(const struct lu_env *env,
 
 
 static inline int dt_write_commit(const struct lu_env *env,
 
 
 static inline int dt_write_commit(const struct lu_env *env,

-                                  struct dt_object *d, struct niobuf_local *lnb,
-                                  int n, struct thandle *th)
+                                 struct dt_object *d, struct niobuf_local *lnb,
+                                 int n, struct thandle *th, __u64 size)

 {
 {

-        LASSERT(d);
-        LASSERT(d->do_body_ops);
-        LASSERT(d->do_body_ops->dbo_write_commit);
-        return d->do_body_ops->dbo_write_commit(env, d, lnb, n, th);
+       LASSERT(d);
+       LASSERT(d->do_body_ops);
+       LASSERT(d->do_body_ops->dbo_write_commit);
+       return d->do_body_ops->dbo_write_commit(env, d, lnb, n, th, size);

 }
 
 static inline int dt_read_prep(const struct lu_env *env, struct dt_object *d,
 }
 
 static inline int dt_read_prep(const struct lu_env *env, struct dt_object *d,

diff --git a/lustre/mdt/mdt_io.c b/lustre/mdt/mdt_io.c
index 75b6f81..e67b73b 100644 (file)
--- a/lustre/mdt/mdt_io.c
+++ b/lustre/mdt/mdt_io.c
@@ -656,7 +656,7 @@ retry:
                GOTO(out_stop, rc);
 
        dt_write_lock(env, dob, 0);
                GOTO(out_stop, rc);
 
        dt_write_lock(env, dob, 0);

-       rc = dt_write_commit(env, dob, lnb, niocount, th);
+       rc = dt_write_commit(env, dob, lnb, niocount, th, 0);

        if (rc)
                GOTO(unlock, rc);
 
        if (rc)
                GOTO(unlock, rc);
 

diff --git a/lustre/ofd/ofd_io.c b/lustre/ofd/ofd_io.c
index 9305d8e..f836194 100644 (file)
--- a/lustre/ofd/ofd_io.c
+++ b/lustre/ofd/ofd_io.c
@@ -1310,7 +1310,7 @@ retry:
                GOTO(out_unlock, rc = -ENOENT);
 
        if (likely(!fake_write)) {
                GOTO(out_unlock, rc = -ENOENT);
 
        if (likely(!fake_write)) {

-               rc = dt_write_commit(env, o, lnb, niocount, th);
+               rc = dt_write_commit(env, o, lnb, niocount, th, oa->o_size);

                if (rc)
                        GOTO(out_unlock, rc);
        }
                if (rc)
                        GOTO(out_unlock, rc);
        }

diff --git a/lustre/osc/osc_request.c b/lustre/osc/osc_request.c
index cb53cc6..e87b4f0 100644 (file)
--- a/lustre/osc/osc_request.c
+++ b/lustre/osc/osc_request.c
@@ -1438,10 +1438,17 @@ retry_encrypt:
                                ptlrpc_request_free(req);
                                RETURN(rc);
                        }
                                ptlrpc_request_free(req);
                                RETURN(rc);
                        }

+                       pg->pg = data_page;
+                       /* there should be no gap in the middle of page array */
+                       if (i == page_count - 1) {
+                               struct osc_async_page *oap = brw_page2oap(pg);
+
+                               oa->o_size = oap->oap_count +
+                                       oap->oap_obj_off + oap->oap_page_off;
+                       }

                        /* len is forced to PAGE_SIZE, and poff to 0
                         * so store the old, clear text info
                         */
                        /* len is forced to PAGE_SIZE, and poff to 0
                         * so store the old, clear text info
                         */

-                       pg->pg = data_page;

                        pg->bp_count_diff = PAGE_SIZE - pg->count;
                        pg->count = PAGE_SIZE;
                        pg->bp_off_diff = pg->off & ~PAGE_MASK;
                        pg->bp_count_diff = PAGE_SIZE - pg->count;
                        pg->count = PAGE_SIZE;
                        pg->bp_off_diff = pg->off & ~PAGE_MASK;

diff --git a/lustre/osd-ldiskfs/osd_io.c b/lustre/osd-ldiskfs/osd_io.c
index 88b4c35..961e10d 100644 (file)
--- a/lustre/osd-ldiskfs/osd_io.c
+++ b/lustre/osd-ldiskfs/osd_io.c
@@ -1221,26 +1221,31 @@ static int osd_declare_write_commit(const struct lu_env *env,
 
 /* Check if a block is allocated or not */
 static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
 
 /* Check if a block is allocated or not */
 static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,

-                            struct niobuf_local *lnb, int npages,
-                            struct thandle *thandle)
+                           struct niobuf_local *lnb, int npages,
+                           struct thandle *thandle, __u64 user_size)

 {
 {

-        struct osd_thread_info *oti = osd_oti_get(env);
-        struct osd_iobuf *iobuf = &oti->oti_iobuf;
-        struct inode *inode = osd_dt_obj(dt)->oo_inode;
-        struct osd_device  *osd = osd_obj2dev(osd_dt_obj(dt));
-        loff_t isize;
-        int rc = 0, i;
+       struct osd_thread_info *oti = osd_oti_get(env);
+       struct osd_iobuf *iobuf = &oti->oti_iobuf;
+       struct inode *inode = osd_dt_obj(dt)->oo_inode;
+       struct osd_device  *osd = osd_obj2dev(osd_dt_obj(dt));
+       loff_t disk_size;
+       int rc = 0, i;

 
 

-        LASSERT(inode);
+       LASSERT(inode);

 
        rc = osd_init_iobuf(osd, iobuf, 1, npages);
        if (unlikely(rc != 0))
                RETURN(rc);
 
 
        rc = osd_init_iobuf(osd, iobuf, 1, npages);
        if (unlikely(rc != 0))
                RETURN(rc);
 

-       isize = i_size_read(inode);
+       disk_size = i_size_read(inode);
+       /* if disk_size is already bigger than specified user_size,
+        * ignore user_size
+        */
+       if (disk_size > user_size)
+               user_size = 0;

        dquot_initialize(inode);
 
        dquot_initialize(inode);
 

-        for (i = 0; i < npages; i++) {
+       for (i = 0; i < npages; i++) {

                if (lnb[i].lnb_rc == -ENOSPC &&
                    (lnb[i].lnb_flags & OBD_BRW_MAPPED)) {
                        /* Allow the write to proceed if overwriting an
                if (lnb[i].lnb_rc == -ENOSPC &&
                    (lnb[i].lnb_flags & OBD_BRW_MAPPED)) {
                        /* Allow the write to proceed if overwriting an


@@ -1260,8 +1265,8 @@ static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
                LASSERT(PageLocked(lnb[i].lnb_page));
                LASSERT(!PageWriteback(lnb[i].lnb_page));
 
                LASSERT(PageLocked(lnb[i].lnb_page));
                LASSERT(!PageWriteback(lnb[i].lnb_page));
 

-               if (lnb[i].lnb_file_offset + lnb[i].lnb_len > isize)
-                       isize = lnb[i].lnb_file_offset + lnb[i].lnb_len;
+               if (lnb[i].lnb_file_offset + lnb[i].lnb_len > disk_size)
+                       disk_size = lnb[i].lnb_file_offset + lnb[i].lnb_len;

 
                /*
                 * Since write and truncate are serialized by oo_sem, even
 
                /*
                 * Since write and truncate are serialized by oo_sem, even


@@ -1273,26 +1278,29 @@ static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
                SetPageUptodate(lnb[i].lnb_page);
 
                osd_iobuf_add_page(iobuf, &lnb[i]);
                SetPageUptodate(lnb[i].lnb_page);
 
                osd_iobuf_add_page(iobuf, &lnb[i]);

-        }
+       }
+       /* if file has grown, take user_size into account */
+       if (user_size && disk_size > user_size)
+               disk_size = user_size;

 
        osd_trans_exec_op(env, thandle, OSD_OT_WRITE);
 
 
        osd_trans_exec_op(env, thandle, OSD_OT_WRITE);
 

-        if (OBD_FAIL_CHECK(OBD_FAIL_OST_MAPBLK_ENOSPC)) {
-                rc = -ENOSPC;
-        } else if (iobuf->dr_npages > 0) {
+       if (OBD_FAIL_CHECK(OBD_FAIL_OST_MAPBLK_ENOSPC)) {
+               rc = -ENOSPC;
+       } else if (iobuf->dr_npages > 0) {

                rc = osd_ldiskfs_map_inode_pages(inode, iobuf->dr_pages,
                                                 iobuf->dr_npages,
                                                 iobuf->dr_blocks, 1);
                rc = osd_ldiskfs_map_inode_pages(inode, iobuf->dr_pages,
                                                 iobuf->dr_npages,
                                                 iobuf->dr_blocks, 1);

-        } else {
-                /* no pages to write, no transno is needed */
-                thandle->th_local = 1;
-        }
+       } else {
+               /* no pages to write, no transno is needed */
+               thandle->th_local = 1;
+       }

 
        if (likely(rc == 0)) {
                spin_lock(&inode->i_lock);
 
        if (likely(rc == 0)) {
                spin_lock(&inode->i_lock);

-               if (isize > i_size_read(inode)) {
-                       i_size_write(inode, isize);
-                       LDISKFS_I(inode)->i_disksize = isize;
+               if (disk_size > i_size_read(inode)) {
+                       i_size_write(inode, disk_size);
+                       LDISKFS_I(inode)->i_disksize = disk_size;

                        spin_unlock(&inode->i_lock);
                        osd_dirty_inode(inode, I_DIRTY_DATASYNC);
                } else {
                        spin_unlock(&inode->i_lock);
                        osd_dirty_inode(inode, I_DIRTY_DATASYNC);
                } else {

diff --git a/lustre/osd-zfs/osd_io.c b/lustre/osd-zfs/osd_io.c
index 59902bd..f2a72b6 100644 (file)
--- a/lustre/osd-zfs/osd_io.c
+++ b/lustre/osd-zfs/osd_io.c
@@ -818,7 +818,7 @@ static void osd_evict_dbufs_after_write(struct osd_object *obj,
 
 static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
                        struct niobuf_local *lnb, int npages,
 
 static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
                        struct niobuf_local *lnb, int npages,

-                       struct thandle *th)
+                       struct thandle *th, __u64 user_size)

 {
        struct osd_object  *obj  = osd_dt_obj(dt);
        struct osd_device  *osd = osd_obj2dev(obj);
 {
        struct osd_object  *obj  = osd_dt_obj(dt);
        struct osd_device  *osd = osd_obj2dev(obj);


@@ -847,6 +847,12 @@ static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
        if (OBD_FAIL_CHECK(OBD_FAIL_OST_MAPBLK_ENOSPC))
                RETURN(-ENOSPC);
 
        if (OBD_FAIL_CHECK(OBD_FAIL_OST_MAPBLK_ENOSPC))
                RETURN(-ENOSPC);
 

+       /* if la_size is already bigger than specified user_size,
+        * ignore user_size
+        */
+       if (obj->oo_attr.la_size > user_size)
+               user_size = 0;
+

        /* LU-8791: take oo_guard to avoid the deadlock that changing block
         * size and assigning arcbuf take place at the same time.
         *
        /* LU-8791: take oo_guard to avoid the deadlock that changing block
         * size and assigning arcbuf take place at the same time.
         *


@@ -948,6 +954,9 @@ static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
                RETURN(0);
        }
 
                RETURN(0);
        }
 

+       /* if file has grown, take user_size into account */
+       if (user_size && new_size > user_size)
+               new_size = user_size;

        write_lock(&obj->oo_attr_lock);
        if (obj->oo_attr.la_size < new_size) {
                obj->oo_attr.la_size = new_size;
        write_lock(&obj->oo_attr_lock);
        if (obj->oo_attr.la_size < new_size) {
                obj->oo_attr.la_size = new_size;