pump: cleanup xattrs on both commit and abort path.

This change makes glusterd to send a setxattr command
for replace-brick commit operation similar to abort.

Earlier we could commit even before the 'migration'
of data was complete, with this change we fail that
operation.

Signed-off-by: Krishnan Parthasarathi <kp@gluster.com>
Signed-off-by: Anand Avati <avati@gluster.com>

BUG: 3033 (Changes to replace-brick and syntask interface.)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=3033

4 files changed, 207 insertions, 4 deletions

diff --git a/xlators/cluster/afr/src/pump.c b/xlators/cluster/afr/src/pump.c
index 5497bddff2b..76280ea8120 100644
--- a/xlators/cluster/afr/src/pump.c
+++ b/xlators/cluster/afr/src/pump.c
@@ -581,6 +581,40 @@ pump_update_resume_path (xlator_t *this)
         return 0;
 }
 
+static int32_t
+pump_xattr_cleaner (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno)
+{
+        afr_private_t  *priv      = NULL;
+        pump_private_t *pump_priv = NULL;
+        loc_t           loc       = {0};
+        int             i         = 0;
+        int             ret       = 0;
+        int             source    = 0;
+        int             sink      = 1;
+
+        priv      = this->private;
+        pump_priv = priv->pump_private;
+
+        build_root_loc (priv->root_inode, &loc);
+
+        ret = syncop_removexattr (priv->children[source], &loc,
+                                          PUMP_PATH);
+
+        ret = syncop_removexattr (priv->children[sink], &loc,
+                                  PUMP_SINK_COMPLETE);
+
+        for (i = 0; i < priv->child_count; i++) {
+                ret = syncop_removexattr (priv->children[i], &loc,
+                                          PUMP_SOURCE_COMPLETE);
+                if (ret)
+                        gf_log (this->name, GF_LOG_DEBUG, "removexattr "
+                                "failed with %s", strerror (errno));
+        }
+
+        return pump_command_reply (frame, this);
+}
+
 static int
 pump_complete_migration (xlator_t *this)
 {
@@ -624,6 +658,11 @@ pump_complete_migration (xlator_t *this)
                 }
 
                 pump_save_path (this, "/");
+
+        } else if (state == PUMP_STATE_ABORT) {
+                gf_log (this->name, GF_LOG_DEBUG, "Starting cleanup "
+                        "of pump internal xattrs");
+                call_resume (pump_priv->cleaner);
         }
 
         return 0;
@@ -1107,12 +1146,73 @@ out:
 	return 0;
 }
 
+static int
+pump_cleanup_helper (void *data) {
+        call_frame_t *frame = data;
+
+        pump_xattr_cleaner (frame, 0, frame->this, 0, 0);
+
+        return 0;
+}
+
+static int
+pump_cleanup_done (int ret, call_frame_t *sync_frame, void *data)
+{
+        STACK_DESTROY (sync_frame->root);
+
+        return 0;
+}
+
+int
+pump_execute_commit (call_frame_t *frame, xlator_t *this)
+{
+        afr_private_t  *priv       = NULL;
+        pump_private_t *pump_priv  = NULL;
+        afr_local_t    *local      = NULL;
+        call_frame_t   *sync_frame = NULL;
+        int             ret        = 0;
+
+        priv      = this->private;
+        pump_priv = priv->pump_private;
+        local     = frame->local;
+
+
+        LOCK (&pump_priv->resume_path_lock);
+        {
+                pump_priv->number_files_pumped = 0;
+                pump_priv->current_file[0] = '\0';
+        }
+        UNLOCK (&pump_priv->resume_path_lock);
+
+        local->op_ret = 0;
+        if (pump_priv->pump_finished) {
+                pump_change_state (this, PUMP_STATE_COMMIT);
+                sync_frame = create_frame (this, this->ctx->pool);
+                ret = synctask_new (pump_priv->env, pump_cleanup_helper,
+                                    pump_cleanup_done, sync_frame, frame);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_DEBUG, "Couldn't create "
+                                "synctask for cleaning up xattrs.");
+                }
+
+        } else {
+                gf_log (this->name, GF_LOG_ERROR, "Commit can't proceed. "
+                        "Migration in progress");
+                local->op_ret = -1;
+                local->op_errno = EINPROGRESS;
+                pump_command_reply (frame, this);
+        }
+
+        return 0;
+}
 int
 pump_execute_abort (call_frame_t *frame, xlator_t *this)
 {
-        afr_private_t  *priv      = NULL;
-        pump_private_t *pump_priv = NULL;
-        afr_local_t    *local     = NULL;
+        afr_private_t  *priv       = NULL;
+        pump_private_t *pump_priv  = NULL;
+        afr_local_t    *local      = NULL;
+        call_frame_t   *sync_frame = NULL;
+        int             ret        = 0;
 
         priv      = this->private;
         pump_priv = priv->pump_private;
@@ -1128,7 +1228,20 @@ pump_execute_abort (call_frame_t *frame, xlator_t *this)
         UNLOCK (&pump_priv->resume_path_lock);
 
         local->op_ret = 0;
-        pump_command_reply (frame, this);
+        if (pump_priv->pump_finished) {
+                sync_frame = create_frame (this, this->ctx->pool);
+                ret = synctask_new (pump_priv->env, pump_cleanup_helper,
+                                    pump_cleanup_done, sync_frame, frame);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_DEBUG, "Couldn't create "
+                                "synctask for cleaning up xattrs.");
+                }
+
+        } else {
+                pump_priv->cleaner = fop_setxattr_cbk_stub (frame,
+                                                            pump_xattr_cleaner,
+                                                            0, 0);
+        }
 
         return 0;
 }
@@ -1182,6 +1295,30 @@ out:
 }
 
 gf_boolean_t
+pump_command_commit (xlator_t *this, dict_t *dict)
+{
+        char *cmd = NULL;
+        int dict_ret = -1;
+        int ret = _gf_true;
+
+        dict_ret = dict_get_str (dict, PUMP_CMD_COMMIT, &cmd);
+        if (dict_ret < 0) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "Not a pump commit command");
+                ret = _gf_false;
+                goto out;
+        }
+
+        gf_log (this->name, GF_LOG_DEBUG,
+                "Hit a pump command - commit");
+        ret = _gf_true;
+
+out:
+        return ret;
+
+}
+
+gf_boolean_t
 pump_command_abort (xlator_t *this, dict_t *dict)
 {
         char *cmd = NULL;
@@ -1596,6 +1733,11 @@ pump_parse_command (call_frame_t *frame, xlator_t *this,
                 frame->local = local;
                 local->dict = dict_ref (dict);
                 ret = pump_execute_abort (frame, this);
+
+        } else if (pump_command_commit (this, dict)) {
+                frame->local = local;
+                local->dict = dict_ref (dict);
+                ret = pump_execute_commit (frame, this);
         }
         return ret;
 }
diff --git a/xlators/cluster/afr/src/pump.h b/xlators/cluster/afr/src/pump.h
index a46f9d7a542..394e6eab753 100644
--- a/xlators/cluster/afr/src/pump.h
+++ b/xlators/cluster/afr/src/pump.h
@@ -34,6 +34,7 @@
 #define IS_ENTRY_PARENT(entry) (!strcmp (entry, ".."))
 
 #define PUMP_CMD_START  "trusted.glusterfs.pump.start"
+#define PUMP_CMD_COMMIT "trusted.glusterfs.pump.commit"
 #define PUMP_CMD_ABORT  "trusted.glusterfs.pump.abort"
 #define PUMP_CMD_PAUSE  "trusted.glusterfs.pump.pause"
 #define PUMP_CMD_STATUS "trusted.glusterfs.pump.status"
@@ -51,6 +52,7 @@ typedef enum {
         PUMP_STATE_RESUME,              /* Pump is resuming from a previous pause */
         PUMP_STATE_PAUSE,               /* Pump is paused */
         PUMP_STATE_ABORT,               /* Pump is aborted */
+        PUMP_STATE_COMMIT,              /* Pump is commited */
 } pump_state_t;
 
 typedef struct _pump_private {
@@ -64,6 +66,7 @@ typedef struct _pump_private {
         gf_boolean_t pump_finished;     /* Boolean to indicate pump termination */
         char pump_start_pending;        /* Boolean to mark start pending until
                                            CHILD_UP */
+        call_stub_t *cleaner;
 } pump_private_t;
 
 void
diff --git a/xlators/mgmt/glusterd/src/glusterd-op-sm.c b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
index a802149212e..6df894ca740 100644
--- a/xlators/mgmt/glusterd/src/glusterd-op-sm.c
+++ b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
@@ -3475,6 +3475,53 @@ rb_kill_destination_brick (glusterd_volinfo_t *volinfo,
 }
 
 static int
+rb_do_operation_commit (glusterd_volinfo_t *volinfo,
+                       glusterd_brickinfo_t *src_brickinfo,
+                       glusterd_brickinfo_t *dst_brickinfo)
+{
+        int ret     = -1;
+        int cmd_ret = -1;
+
+        gf_log ("", GF_LOG_DEBUG,
+                "replace-brick sending commit xattr");
+
+        ret = rb_spawn_maintenance_client (volinfo, src_brickinfo);
+        if (ret) {
+                gf_log ("", GF_LOG_DEBUG,
+                        "Could not spawn maintenance "
+                        "client");
+                goto out;
+        }
+
+	gf_log ("", GF_LOG_DEBUG,
+		"mounted the replace brick client");
+
+        cmd_ret = rb_send_xattr_command (volinfo, src_brickinfo,
+                                     dst_brickinfo, RB_PUMP_COMMIT_CMD,
+                                     "jargon");
+        if (cmd_ret) {
+                gf_log ("", GF_LOG_DEBUG,
+                        "Failed to send command to pump");
+	}
+
+        ret = rb_destroy_maintenance_client (volinfo, src_brickinfo);
+        if (ret) {
+                gf_log ("", GF_LOG_DEBUG,
+                        "Failed to destroy maintenance "
+                        "client");
+                goto out;
+        }
+
+	gf_log ("", GF_LOG_DEBUG,
+		"unmounted the replace brick client");
+
+        ret = 0;
+
+out:
+        return cmd_ret || ret;
+}
+
+static int
 rb_do_operation_abort (glusterd_volinfo_t *volinfo,
                        glusterd_brickinfo_t *src_brickinfo,
                        glusterd_brickinfo_t *dst_brickinfo)
@@ -3894,6 +3941,16 @@ glusterd_op_replace_brick (dict_t *dict, dict_t *rsp_dict)
         case GF_REPLACE_OP_COMMIT:
         case GF_REPLACE_OP_COMMIT_FORCE:
         {
+                ctx = glusterd_op_get_ctx (GD_OP_REPLACE_BRICK);
+                if (ctx) {
+                        ret = rb_do_operation_commit (volinfo, src_brickinfo, dst_brickinfo);
+                        if (ret) {
+                                gf_log ("", GF_LOG_ERROR,
+                                        "Commit operation failed");
+                                goto out;
+                        }
+                }
+
                 ret = dict_set_int32 (volinfo->dict, "enable-pump", 0);
                 gf_log ("", GF_LOG_DEBUG,
                         "Received commit - will be adding dst brick and "
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index 2ddd830f847..b1027c1714c 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -245,6 +245,7 @@ enum glusterd_vol_comp_status_ {
 /*All definitions related to replace brick */
 #define RB_PUMP_START_CMD       "trusted.glusterfs.pump.start"
 #define RB_PUMP_PAUSE_CMD       "trusted.glusterfs.pump.pause"
+#define RB_PUMP_COMMIT_CMD      "trusted.glusterfs.pump.commit"
 #define RB_PUMP_ABORT_CMD       "trusted.glusterfs.pump.abort"
 #define RB_PUMP_STATUS_CMD      "trusted.glusterfs.pump.status"
 #define RB_CLIENT_MOUNTPOINT    "rb_mount"