X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_hsm_cdt_client.c;h=22df94c116ea03e69d61ab09dc1ed728ee120c5f;hb=2e2c8d59cfdb6d7078f9627ca970920059fcf90f;hp=af2ddf491bdd574dc54ae835856a806d28477426;hpb=adce970c84628879840c66e263fdb1f8506a20f2;p=fs%2Flustre-release.git

diff --git a/lustre/mdt/mdt_hsm_cdt_client.c b/lustre/mdt/mdt_hsm_cdt_client.c
index af2ddf4..22df94c 100644
--- a/lustre/mdt/mdt_hsm_cdt_client.c
+++ b/lustre/mdt/mdt_hsm_cdt_client.c
@@ -23,6 +23,7 @@
  * (C) Copyright 2012 Commissariat a l'energie atomique et aux energies
  *     alternatives
  *
+ * Copyright (c) 2013, 2014, Intel Corporation.
  */
 /*
  * lustre/mdt/mdt_hsm_cdt_client.c
@@ -39,7 +40,6 @@
 #include <lustre_net.h>
 #include <lustre_export.h>
 #include <obd.h>
-#include <obd_lov.h>
 #include <lprocfs_status.h>
 #include <lustre_log.h>
 #include "mdt_internal.h"
@@ -77,12 +77,12 @@ static int hsm_find_compatible_cb(const struct lu_env *env,
 	hcdcb = data;
 	/* a compatible request must be WAITING or STARTED
 	 * and not a cancel */
-	if (((larr->arr_status != ARS_WAITING) &&
-	     (larr->arr_status != ARS_STARTED)) ||
-	    (larr->arr_hai.hai_action == HSMA_CANCEL))
+	if ((larr->arr_status != ARS_WAITING &&
+	     larr->arr_status != ARS_STARTED) ||
+	    larr->arr_hai.hai_action == HSMA_CANCEL)
 		RETURN(0);
 
-	hai = hai_zero(hcdcb->hal);
+	hai = hai_first(hcdcb->hal);
 	for (i = 0; i < hcdcb->hal->hal_count; i++, hai = hai_next(hai)) {
 		/* if request is a CANCEL:
 		 * if cookie set in the request, there is no need to find a
@@ -92,7 +92,7 @@ static int hsm_find_compatible_cb(const struct lu_env *env,
 		 * if the caller sets the cookie, we assume he also sets the
 		 * arr_archive_id
 		 */
-		if ((hai->hai_action == HSMA_CANCEL) && (hai->hai_cookie != 0))
+		if (hai->hai_action == HSMA_CANCEL && hai->hai_cookie != 0)
 			continue;
 
 		if (!lu_fid_eq(&hai->hai_fid, &larr->arr_hai.hai_fid))
@@ -110,8 +110,8 @@ static int hsm_find_compatible_cb(const struct lu_env *env,
 		 */
 		hai->hai_cookie = larr->arr_hai.hai_cookie;
 		/* we read the archive number from the request we cancel */
-		if ((hai->hai_action == HSMA_CANCEL) &&
-		    (hcdcb->hal->hal_archive_id == 0))
+		if (hai->hai_action == HSMA_CANCEL &&
+		    hcdcb->hal->hal_archive_id == 0)
 			hcdcb->hal->hal_archive_id = larr->arr_archive_id;
 	}
 	RETURN(0);
@@ -136,13 +136,13 @@ static int hsm_find_compatible(const struct lu_env *env, struct mdt_device *mdt,
 	ENTRY;
 
 	ok_cnt = 0;
-	hai = hai_zero(hal);
+	hai = hai_first(hal);
 	for (i = 0; i < hal->hal_count; i++, hai = hai_next(hai)) {
 		/* in a cancel request hai_cookie may be set by caller to
 		 * show the request to be canceled
 		 * if not we need to search by FID
 		 */
-		if ((hai->hai_action == HSMA_CANCEL) && (hai->hai_cookie != 0))
+		if (hai->hai_action == HSMA_CANCEL && hai->hai_cookie != 0)
 			ok_cnt++;
 		else
 			hai->hai_cookie = 0;
@@ -181,7 +181,7 @@ static bool hsm_action_is_needed(struct hsm_action_item *hai, int hal_an,
 	hsm_flags = hsm->mh_flags;
 	switch (hai->hai_action) {
 	case HSMA_ARCHIVE:
-		if ((hsm_flags & HS_DIRTY) || !(hsm_flags & HS_ARCHIVED))
+		if (hsm_flags & HS_DIRTY || !(hsm_flags & HS_ARCHIVED))
 			is_needed = true;
 		break;
 	case HSMA_RESTORE:
@@ -200,8 +200,8 @@ static bool hsm_action_is_needed(struct hsm_action_item *hai, int hal_an,
 		is_needed = true;
 		break;
 	}
-	CDEBUG(D_HSM, "fid="DFID" action=%s rq_flags="LPX64
-		      " extent="LPX64"-"LPX64" hsm_flags=%X %s\n",
+	CDEBUG(D_HSM, "fid="DFID" action=%s rq_flags=%#llx"
+		      " extent=%#llx-%#llx hsm_flags=%X %s\n",
 		      PFID(&hai->hai_fid),
 		      hsm_copytool_action2name(hai->hai_action), rq_flags,
 		      hai->hai_extent.offset, hai->hai_extent.length,
@@ -227,7 +227,7 @@ static bool hal_is_sane(struct hsm_action_list *hal)
 	if (hal->hal_count == 0)
 		RETURN(false);
 
-	hai = hai_zero(hal);
+	hai = hai_first(hal);
 	for (i = 0; i < hal->hal_count; i++, hai = hai_next(hai)) {
 		if (!fid_is_sane(&hai->hai_fid))
 			RETURN(false);
@@ -245,6 +245,42 @@ static bool hal_is_sane(struct hsm_action_list *hal)
 	RETURN(true);
 }
 
+static int
+hsm_action_permission(struct mdt_thread_info *mti,
+		      struct mdt_object *obj,
+		      enum hsm_copytool_action hsma)
+{
+	struct coordinator *cdt = &mti->mti_mdt->mdt_coordinator;
+	struct lu_ucred *uc = mdt_ucred(mti);
+	struct md_attr *ma = &mti->mti_attr;
+	const __u64 *mask;
+	int rc;
+	ENTRY;
+
+	if (hsma != HSMA_RESTORE && mdt_rdonly(mti->mti_exp))
+		RETURN(-EROFS);
+
+	if (md_capable(uc, CFS_CAP_SYS_ADMIN))
+		RETURN(0);
+
+	ma->ma_need = MA_INODE;
+	rc = mdt_attr_get_complex(mti, obj, ma);
+	if (rc < 0)
+		RETURN(rc);
+
+	if (uc->uc_fsuid == ma->ma_attr.la_uid)
+		mask = &cdt->cdt_user_request_mask;
+	else if (lustre_in_group_p(uc, ma->ma_attr.la_gid))
+		mask = &cdt->cdt_group_request_mask;
+	else
+		mask = &cdt->cdt_other_request_mask;
+
+	if (!(0 <= hsma && hsma < 8 * sizeof(*mask)))
+		RETURN(-EINVAL);
+
+	RETURN(*mask & (1UL << hsma) ? 0 : -EPERM);
+}
+
 /*
  * Coordinator external API
  */
@@ -253,13 +289,12 @@ static bool hal_is_sane(struct hsm_action_list *hal)
  * register a list of requests
  * \param mti [IN]
  * \param hal [IN] list of requests
- * \param compound_id [OUT] id of the compound request
  * \retval 0 success
  * \retval -ve failure
  * in case of restore, caller must hold layout lock
  */
 int mdt_hsm_add_actions(struct mdt_thread_info *mti,
-			struct hsm_action_list *hal, __u64 *compound_id)
+			struct hsm_action_list *hal)
 {
 	struct mdt_device	*mdt = mti->mti_mdt;
 	struct coordinator	*cdt = &mdt->mdt_coordinator;
@@ -268,6 +303,7 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 	int			 rc = 0, i;
 	struct md_hsm		 mh;
 	bool			 is_restore = false;
+	__u64			 compound_id;
 	ENTRY;
 
 	/* no coordinator started, so we cannot serve requests */
@@ -277,7 +313,7 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 	if (!hal_is_sane(hal))
 		RETURN(-EINVAL);
 
-	*compound_id = cfs_atomic_inc_return(&cdt->cdt_compound_id);
+	compound_id = atomic_inc_return(&cdt->cdt_compound_id);
 
 	/* search for compatible request, if found hai_cookie is set
 	 * to the request cookie
@@ -287,7 +323,7 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 	if (rc)
 		GOTO(out, rc);
 
-	hai = hai_zero(hal);
+	hai = hai_first(hal);
 	for (i = 0; i < hal->hal_count; i++, hai = hai_next(hai)) {
 		int archive_id;
 		__u64 flags;
@@ -310,10 +346,10 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 		 * do not record
 		 */
 		/* redundant case */
-		if ((hai->hai_action != HSMA_CANCEL) && (hai->hai_cookie != 0))
+		if (hai->hai_action != HSMA_CANCEL && hai->hai_cookie != 0)
 			continue;
 		/* cancel nothing case */
-		if ((hai->hai_action == HSMA_CANCEL) && (hai->hai_cookie == 0))
+		if (hai->hai_action == HSMA_CANCEL && hai->hai_cookie == 0)
 			continue;
 
 		/* new request or cancel request
@@ -321,20 +357,30 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 		 * if restore, we take the layout lock
 		 */
 
-		/* if action is cancel, also no need to check */
-		if (hai->hai_action == HSMA_CANCEL)
-			goto record;
-
-		/* get HSM attributes */
-		obj = mdt_hsm_get_md_hsm(mti, &hai->hai_fid, &mh, NULL);
+		/* Get HSM attributes and check permissions. */
+		obj = mdt_hsm_get_md_hsm(mti, &hai->hai_fid, &mh);
 		if (IS_ERR(obj)) {
-			/* in case of archive remove, Lustre file
-			 * is not mandatory */
-			if (hai->hai_action == HSMA_REMOVE)
+			/* In case of REMOVE and CANCEL a Lustre file
+			 * is not mandatory, but restrict this
+			 * exception to admins. */
+			if (md_capable(mdt_ucred(mti), CFS_CAP_SYS_ADMIN) &&
+			    (hai->hai_action == HSMA_REMOVE ||
+			     hai->hai_action == HSMA_CANCEL))
 				goto record;
-			GOTO(out, rc = PTR_ERR(obj));
+			else
+				GOTO(out, rc = PTR_ERR(obj));
 		}
 
+		rc = hsm_action_permission(mti, obj, hai->hai_action);
+		mdt_object_put(mti->mti_env, obj);
+
+		if (rc < 0)
+			GOTO(out, rc);
+
+		/* if action is cancel, also no need to check */
+		if (hai->hai_action == HSMA_CANCEL)
+			goto record;
+
 		/* Check if an action is needed, compare request
 		 * and HSM flags status */
 		if (!hsm_action_is_needed(hai, archive_id, flags, &mh))
@@ -348,6 +394,7 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 
 		/* for cancel archive number is taken from canceled request
 		 * for other request, we take from lma if not specified,
+		 * or we use the default if none found in lma
 		 * this works also for archive because the default value is 0
 		 * /!\ there is a side effect: in case of restore on multiple
 		 * files which are in different backend, the initial compound
@@ -355,53 +402,56 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 		 * warranty an agent can serve any combinaison of archive
 		 * backend
 		 */
-		if ((hai->hai_action != HSMA_CANCEL) &&
-		    (archive_id == 0))
-			archive_id = mh.mh_arch_id;
+		if (hai->hai_action != HSMA_CANCEL && archive_id == 0) {
+			if (mh.mh_arch_id != 0)
+				archive_id = mh.mh_arch_id;
+			else
+				archive_id = cdt->cdt_default_archive_id;
+		}
 
 		/* if restore, take an exclusive lock on layout */
 		if (hai->hai_action == HSMA_RESTORE) {
-			struct cdt_restore_handle	*crh;
-			struct mdt_object		*child;
+			struct cdt_restore_handle *crh;
+
+			/* in V1 only whole file is supported. */
+			if (hai->hai_extent.offset != 0)
+				GOTO(out, rc = -EPROTO);
 
-			OBD_ALLOC_PTR(crh);
+			OBD_SLAB_ALLOC_PTR(crh, mdt_hsm_cdt_kmem);
 			if (crh == NULL)
 				GOTO(out, rc = -ENOMEM);
 
 			crh->crh_fid = hai->hai_fid;
-			/* in V1 only whole file is supported
-			crh->extent.start = hai->hai_extent.offset;
-			crh->extent.end = hai->hai_extent.offset +
-					    hai->hai_extent.length;
-			 */
+			/* in V1 only whole file is supported. However the
+			 * restore may be due to truncate. */
 			crh->crh_extent.start = 0;
-			crh->crh_extent.end = OBD_OBJECT_EOF;
+			crh->crh_extent.end = hai->hai_extent.length;
 
 			mdt_lock_reg_init(&crh->crh_lh, LCK_EX);
-			child = mdt_object_find_lock(mti, &crh->crh_fid,
-						     &crh->crh_lh,
-						     MDS_INODELOCK_LAYOUT);
-			if (IS_ERR(child)) {
-				rc = PTR_ERR(child);
+			obj = mdt_object_find_lock(mti, &crh->crh_fid,
+						   &crh->crh_lh,
+						   MDS_INODELOCK_LAYOUT);
+			if (IS_ERR(obj)) {
+				rc = PTR_ERR(obj);
 				CERROR("%s: cannot take layout lock for "
 				       DFID": rc = %d\n", mdt_obd_name(mdt),
 				       PFID(&crh->crh_fid), rc);
-				OBD_FREE_PTR(crh);
+				OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
 				GOTO(out, rc);
 			}
+
 			/* we choose to not keep a keep a reference
 			 * on the object during the restore time which can be
 			 * very long */
-			mdt_object_put(mti->mti_env, child);
+			mdt_object_put(mti->mti_env, obj);
 
-			down(&cdt->cdt_restore_lock);
-			cfs_list_add_tail(&crh->crh_list,
-					  &cdt->cdt_restore_hdl);
-			up(&cdt->cdt_restore_lock);
+			mutex_lock(&cdt->cdt_restore_lock);
+			list_add_tail(&crh->crh_list, &cdt->cdt_restore_hdl);
+			mutex_unlock(&cdt->cdt_restore_lock);
 		}
 record:
 		/* record request */
-		rc = mdt_agent_record_add(mti->mti_env, mdt, *compound_id,
+		rc = mdt_agent_record_add(mti->mti_env, mdt, compound_id,
 					  archive_id, flags, hai);
 		if (rc)
 			GOTO(out, rc);
@@ -411,10 +461,11 @@ record:
 		rc = -ENODATA;
 	else
 		rc = 0;
-	EXIT;
+
+	GOTO(out, rc);
 out:
 	/* if work has been added, wake up coordinator */
-	if ((rc == 0) || (rc == -ENODATA))
+	if (rc == 0 || rc == -ENODATA)
 		mdt_hsm_cdt_wakeup(mdt);
 
 	return rc;
@@ -436,7 +487,7 @@ int mdt_hsm_get_running(struct mdt_thread_info *mti,
 	int			 i;
 	ENTRY;
 
-	hai = hai_zero(hal);
+	hai = hai_first(hal);
 	for (i = 0; i < hal->hal_count; i++, hai = hai_next(hai)) {
 		struct cdt_agent_req *car;
 
@@ -444,7 +495,7 @@ int mdt_hsm_get_running(struct mdt_thread_info *mti,
 			RETURN(-EINVAL);
 
 		car = mdt_cdt_find_request(cdt, 0, &hai->hai_fid);
-		if (IS_ERR(car)) {
+		if (car == NULL) {
 			hai->hai_cookie = 0;
 			hai->hai_action = HSMA_NONE;
 		} else {
@@ -469,7 +520,6 @@ bool mdt_hsm_restore_is_running(struct mdt_thread_info *mti,
 {
 	struct mdt_device		*mdt = mti->mti_mdt;
 	struct coordinator		*cdt = &mdt->mdt_coordinator;
-	cfs_list_t			*pos, *tmp;
 	struct cdt_restore_handle	*crh;
 	bool				 rc = false;
 	ENTRY;
@@ -477,15 +527,14 @@ bool mdt_hsm_restore_is_running(struct mdt_thread_info *mti,
 	if (!fid_is_sane(fid))
 		RETURN(rc);
 
-	down(&cdt->cdt_restore_lock);
-	cfs_list_for_each_safe(pos, tmp, &cdt->cdt_restore_hdl) {
-		crh = cfs_list_entry(pos, struct cdt_restore_handle, crh_list);
+	mutex_lock(&cdt->cdt_restore_lock);
+	list_for_each_entry(crh, &cdt->cdt_restore_hdl, crh_list) {
 		if (lu_fid_eq(&crh->crh_fid, fid)) {
 			rc = true;
 			break;
 		}
 	}
-	up(&cdt->cdt_restore_lock);
+	mutex_unlock(&cdt->cdt_restore_lock);
 	RETURN(rc);
 }
 
@@ -505,7 +554,7 @@ int mdt_hsm_get_actions(struct mdt_thread_info *mti,
 	int			 i, rc;
 	ENTRY;
 
-	hai = hai_zero(hal);
+	hai = hai_first(hal);
 	for (i = 0; i < hal->hal_count; i++, hai = hai_next(hai)) {
 		hai->hai_action = HSMA_NONE;
 		if (!fid_is_sane(&hai->hai_fid))
@@ -528,12 +577,12 @@ int mdt_hsm_get_actions(struct mdt_thread_info *mti,
 	 * we may want do give back dynamic informations on the
 	 * running request
 	 */
-	hai = hai_zero(hal);
+	hai = hai_first(hal);
 	for (i = 0; i < hal->hal_count; i++, hai = hai_next(hai)) {
 		struct cdt_agent_req *car;
 
 		car = mdt_cdt_find_request(cdt, hai->hai_cookie, NULL);
-		if (IS_ERR(car)) {
+		if (car == NULL) {
 			hai->hai_cookie = 0;
 		} else {
 			__u64 data_moved;
@@ -550,4 +599,3 @@ int mdt_hsm_get_actions(struct mdt_thread_info *mti,
 
 	RETURN(0);
 }
-