func (ta *TabletActor) multiSnapshot(actionNode *actionnode.ActionNode) error {
	args := actionNode.Args.(*actionnode.MultiSnapshotArgs)

	tablet, err := ta.ts.GetTablet(ta.tabletAlias)
	if err != nil {
		return err
	}
	ki, err := ta.ts.GetKeyspace(tablet.Keyspace)
	if err != nil {
		return err
	}

	if tablet.Type != topo.TYPE_BACKUP {
		return fmt.Errorf("expected backup type, not %v: %v", tablet.Type, ta.tabletAlias)
	}

	filenames, err := ta.mysqld.CreateMultiSnapshot(args.KeyRanges, tablet.DbName(), ki.ShardingColumnName, ki.ShardingColumnType, tablet.Addr(), false, args.Concurrency, args.Tables, args.SkipSlaveRestart, args.MaximumFilesize, ta.hookExtraEnv())
	if err != nil {
		return err
	}

	sr := &actionnode.MultiSnapshotReply{ManifestPaths: filenames}
	if tablet.Parent.Uid == topo.NO_TABLET {
		// If this is a master, this will be the new parent.
		// FIXME(msolomon) this doens't work in hierarchical replication.
		sr.ParentAlias = tablet.Alias
	} else {
		sr.ParentAlias = tablet.Parent
	}
	actionNode.Reply = sr
	return nil
}

func (ta *TabletActor) executeHook(actionNode *actionnode.ActionNode) (err error) {
	// FIXME(msolomon) shouldn't the reply get distilled into an error?
	h := actionNode.Args.(*hook.Hook)
	topotools.ConfigureTabletHook(h, ta.tabletAlias)
	actionNode.Reply = h.Execute()
	return nil
}

// Operate on a backup tablet. Shutdown mysqld and copy the data files aside.
func (ta *TabletActor) snapshot(actionNode *actionnode.ActionNode) error {
	args := actionNode.Args.(*actionnode.SnapshotArgs)

	tablet, err := ta.ts.GetTablet(ta.tabletAlias)
	if err != nil {
		return err
	}

	if tablet.Type != topo.TYPE_BACKUP {
		return fmt.Errorf("expected backup type, not %v: %v", tablet.Type, ta.tabletAlias)
	}

	filename, slaveStartRequired, readOnly, err := ta.mysqld.CreateSnapshot(tablet.DbName(), tablet.Addr(), false, args.Concurrency, args.ServerMode, ta.hookExtraEnv())
	if err != nil {
		return err
	}

	sr := &actionnode.SnapshotReply{ManifestPath: filename, SlaveStartRequired: slaveStartRequired, ReadOnly: readOnly}
	if tablet.Parent.Uid == topo.NO_TABLET {
		// If this is a master, this will be the new parent.
		// FIXME(msolomon) this doesn't work in hierarchical replication.
		sr.ParentAlias = tablet.Alias
	} else {
		sr.ParentAlias = tablet.Parent
	}
	actionNode.Reply = sr
	return nil
}

func (ta *TabletActor) reparentPosition(actionNode *actionnode.ActionNode) error {
	slavePos := actionNode.Args.(*myproto.ReplicationPosition)

	replicationState, waitPosition, timePromoted, err := ta.mysqld.ReparentPosition(slavePos)
	if err != nil {
		return err
	}
	rsd := new(actionnode.RestartSlaveData)
	rsd.ReplicationState = replicationState
	rsd.TimePromoted = timePromoted
	rsd.WaitPosition = waitPosition
	rsd.Parent = ta.tabletAlias
	log.V(6).Infof("reparentPosition %v", rsd.String())
	actionNode.Reply = rsd
	return nil
}

func (ta *TabletActor) applySchema(actionNode *actionnode.ActionNode) error {
	sc := actionNode.Args.(*myproto.SchemaChange)

	// read the tablet to get the dbname
	tablet, err := ta.ts.GetTablet(ta.tabletAlias)
	if err != nil {
		return err
	}

	// and apply the change
	scr, err := ta.mysqld.ApplySchemaChange(tablet.DbName(), sc)
	if err != nil {
		return err
	}
	actionNode.Reply = scr
	return nil
}

func (ta *TabletActor) preflightSchema(actionNode *actionnode.ActionNode) error {
	change := actionNode.Args.(*string)

	// read the tablet to get the dbname
	tablet, err := ta.ts.GetTablet(ta.tabletAlias)
	if err != nil {
		return err
	}

	// and preflight the change
	scr, err := ta.mysqld.PreflightSchemaChange(tablet.DbName(), *change)
	if err != nil {
		return err
	}
	actionNode.Reply = scr
	return nil
}

func (ta *TabletActor) promoteSlave(actionNode *actionnode.ActionNode) error {
	tablet, err := ta.ts.GetTablet(ta.tabletAlias)
	if err != nil {
		return err
	}

	// Perform the action.
	rsd := &actionnode.RestartSlaveData{Parent: tablet.Alias, Force: (tablet.Parent.Uid == topo.NO_TABLET)}
	rsd.ReplicationState, rsd.WaitPosition, rsd.TimePromoted, err = ta.mysqld.PromoteSlave(false, ta.hookExtraEnv())
	if err != nil {
		return err
	}
	log.Infof("PromoteSlave %v", rsd.String())
	actionNode.Reply = rsd

	return updateReplicationGraphForPromotedSlave(ta.ts, tablet)
}

// Write the result of an action into topology server
func StoreActionResponse(ts topo.Server, actionNode *actionnode.ActionNode, actionPath string, actionErr error) error {
	// change our state
	if actionErr != nil {
		// on failure, set an error field on the node
		actionNode.Error = actionErr.Error()
		actionNode.State = actionnode.ACTION_STATE_FAILED
	} else {
		actionNode.Error = ""
		actionNode.State = actionnode.ACTION_STATE_DONE
	}
	actionNode.Pid = 0

	// Write the data first to our action node, then to the log.
	// In the error case, this node will be left behind to debug.
	data := actionNode.ToJson()
	return ts.StoreTabletActionResponse(actionPath, data)
}

func (wr *Wrangler) unlockShard(keyspace, shard string, actionNode *actionnode.ActionNode, lockPath string, actionError error) error {
	// first update the actionNode
	if actionError != nil {
		log.Infof("Unlocking shard %v/%v for action %v with error %v", keyspace, shard, actionNode.Action, actionError)
		actionNode.Error = actionError.Error()
		actionNode.State = actionnode.ACTION_STATE_FAILED
	} else {
		log.Infof("Unlocking shard %v/%v for successful action %v", keyspace, shard, actionNode.Action)
		actionNode.Error = ""
		actionNode.State = actionnode.ACTION_STATE_DONE
	}
	err := wr.ts.UnlockShardForAction(keyspace, shard, lockPath, actionNode.ToJson())
	if actionError != nil {
		if err != nil {
			// this will be masked
			log.Warningf("UnlockShardForAction failed: %v", err)
		}
		return actionError
	}
	return err
}

func (wr *Wrangler) unlockKeyspace(keyspace string, actionNode *actionnode.ActionNode, lockPath string, actionError error) error {
	return actionNode.UnlockKeyspace(wr.ts, keyspace, lockPath, actionError)
}

func (wr *Wrangler) lockKeyspace(keyspace string, actionNode *actionnode.ActionNode) (lockPath string, err error) {
	return actionNode.LockKeyspace(wr.ts, keyspace, wr.lockTimeout, interrupted)
}

func (wr *Wrangler) unlockShard(ctx context.Context, keyspace, shard string, actionNode *actionnode.ActionNode, lockPath string, actionError error) error {
	return actionNode.UnlockShard(ctx, wr.ts, keyspace, shard, lockPath, actionError)
}

func (wr *Wrangler) lockShard(ctx context.Context, keyspace, shard string, actionNode *actionnode.ActionNode) (lockPath string, err error) {
	ctx, cancel := context.WithTimeout(ctx, wr.lockTimeout)
	defer cancel()
	return actionNode.LockShard(ctx, wr.ts, keyspace, shard)
}

func (wr *Wrangler) lockShard(keyspace, shard string, actionNode *actionnode.ActionNode) (lockPath string, err error) {
	log.Infof("Locking shard %v/%v for action %v", keyspace, shard, actionNode.Action)
	return wr.ts.LockShardForAction(keyspace, shard, actionNode.ToJson(), wr.lockTimeout, interrupted)
}

func (wr *Wrangler) lockKeyspace(ctx context.Context, keyspace string, actionNode *actionnode.ActionNode) (lockPath string, err error) {
	return actionNode.LockKeyspace(ctx, wr.ts, keyspace)
}

func (ai *ActionInitiator) writeTabletAction(tabletAlias topo.TabletAlias, node *actionnode.ActionNode) (actionPath string, err error) {
	data := node.SetGuid().ToJson()
	return ai.ts.WriteTabletAction(tabletAlias, data)
}