pause_pauser.go

// @copyright 2022-Present Couchbase, Inc.
//
// Use of this software is governed by the Business Source License included
// in the file licenses/BSL-Couchbase.txt.  As of the Change Date specified
// in that file, in accordance with the Business Source License, use of this
// software will be governed by the Apache License, Version 2.0, included in
// the file licenses/APL2.txt.

package indexer

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"strings"
	"sync"
	"time"

	"github.com/couchbase/cbauth/metakv"
	"github.com/couchbase/indexing/secondary/common"
	"github.com/couchbase/indexing/secondary/logging"
	"github.com/couchbase/indexing/secondary/manager"
)

////////////////////////////////////////////////////////////////////////////////////////////////////
// PauseUploadToken and PauseUploadState
// Used to convey state change information between Pause master and follower nodes.
////////////////////////////////////////////////////////////////////////////////////////////////////

type PauseUploadState byte

const (
	// Posted is the initial state as generated by master. Indicates to followers to start upload work.
	// Followers change state to InProgress.
	PauseUploadTokenPosted PauseUploadState = iota

	// InProgess indicates that followers are actually performing the upload work. In addition, any
	// upload work to be done only by master is also carried out. Once all the work is done, followers
	// change the state to Processed.
	PauseUploadTokenInProgess

	// Processed indicates that for a follower, all the upload work is completed. Master will delete the token
	// and once all followers are done, cleanup is initiated.
	PauseUploadTokenProcessed

	// Error indicates that for a follower, an error was encountered during upload. Master will initiate cleanup.
	PauseUploadTokenError
)

func (s PauseUploadState) String() string {
	switch s {
	case PauseUploadTokenPosted:
		return "PauseUploadTokenPosted"
	case PauseUploadTokenInProgess:
		return "PauseUploadTokenInProgess"
	case PauseUploadTokenProcessed:
		return "PauseUploadTokenProcessed"
	case PauseUploadTokenError:
		return "PauseUploadTokenError"
	}

	return fmt.Sprintf("PauseUploadState-UNKNOWN-STATE-[%v]", byte(s))
}

const PauseUploadTokenTag = "PauseUploadToken"
const PauseUploadTokenPathPrefix = PauseMetakvDir + PauseUploadTokenTag

type PauseUploadToken struct {
	MasterId   string
	FollowerId string
	PauseId    string
	State      PauseUploadState
	BucketName string
	Error      string
}

func newPauseUploadToken(masterUuid, followerUuid, pauseId, bucketName string) (string, *PauseUploadToken, error) {
	put := &PauseUploadToken{
		MasterId:   masterUuid,
		FollowerId: followerUuid,
		PauseId:    pauseId,
		State:      PauseUploadTokenPosted,
		BucketName: bucketName,
	}

	ustr, err := common.NewUUID()
	if err != nil {
		logging.Warnf("newPauseUploadToken: Failed to generate uuid: err[%v]", err)
		return "", nil, err
	}

	putId := fmt.Sprintf("%s%s", PauseUploadTokenTag, ustr.Str())

	return putId, put, nil
}

func decodePauseUploadToken(path string, value []byte) (string, *PauseUploadToken, error) {

	putIdPos := strings.Index(path, PauseUploadTokenTag)
	putId := path[putIdPos:]

	put := &PauseUploadToken{}
	err := json.Unmarshal(value, put)
	if err != nil {
		logging.Errorf("decodePauseUploadToken: Failed to unmarshal value[%s] path[%v]: err[%v]",
			string(value), path, err)
		return "", nil, err
	}

	return putId, put, nil
}

func (put *PauseUploadToken) Clone() *PauseUploadToken {
	put1 := *put
	put2 := put1
	return &put2
}

func setPauseUploadTokenInMetakv(putId string, put *PauseUploadToken) {

	rhCb := func(r int, err error) error {
		if r > 0 {
			logging.Warnf("setPauseUploadTokenInMetakv::rhCb: err[%v], Retrying[%d]", err, r)
		}

		return common.MetakvSet(PauseMetakvDir+putId, put)
	}

	rh := common.NewRetryHelper(10, time.Second, 1, rhCb)
	err := rh.Run()

	if err != nil {
		logging.Fatalf("setPauseUploadTokenInMetakv: Failed to set PauseUploadToken In Meta Storage:" +
			" putId[%v] put[%v] err[%v]", putId, put, err)
		common.CrashOnError(err)
	}
}

////////////////////////////////////////////////////////////////////////////////////////////////////
// Pauser class - Perform the Pause of a given bucket (similar to Rebalancer's role).
// This is used only on the master node of a task_PAUSE task to do the GSI orchestration.
////////////////////////////////////////////////////////////////////////////////////////////////////

// Pauser object holds the state of Pause orchestration
type Pauser struct {
	// nodeDir is "node_<nodeId>/" for this node, where nodeId is the 32-digit hex ID from ns_server
	nodeDir string

	// otherIndexAddrs is "host:port" to all the known Index Service nodes EXCLUDING this one
	otherIndexAddrs []string

	// pauseMgr is the singleton parent of this object
	pauseMgr *PauseServiceManager

	// task is the task_PAUSE task we are executing (protected by task.taskMu). It lives in the
	// pauseMgr.tasks map (protected by pauseMgr.tasksMu). Only the current object should change or
	// delete task at this point, but GetTaskList and other processing may concurrently read it.
	// Thus Pauser needs to write lock task.taskMu for changes but does not need to read lock it.
	task *taskObj

	// Channels used for signalling
	// Used to signal that the PauseUploadTokens have been published.
	waitForTokenPublish chan struct{}
	// Used to signal metakv observer to stop
	metakvCancel chan struct{}

	metakvMutex sync.RWMutex
	wg          sync.WaitGroup

	// Global token associated with this Pause task
	pauseToken *PauseToken

	// in-memory bookkeeping for observed tokens
	masterTokens, followerTokens map[string]*PauseUploadToken

	// lock protecting access to maps like masterTokens and followerTokens
	mu sync.RWMutex

	// For cleanup
	retErr      error
	cleanupOnce sync.Once
}

// RunPauser creates a Pauser instance to execute the given task. It saves a pointer to itself in
// task.pauser (visible to pauseMgr parent) and launches a goroutine for the work.
//
//	pauseMgr - parent object (singleton)
//	task - the task_PAUSE task this object will execute
//	master - true iff this node is the master
func RunPauser(pauseMgr *PauseServiceManager, task *taskObj, master bool, pauseToken *PauseToken) {
	pauser := &Pauser{
		pauseMgr: pauseMgr,
		task:     task,
		nodeDir:  "node_" + string(pauseMgr.genericMgr.nodeInfo.NodeID) + "/",

		waitForTokenPublish: make(chan struct{}),
		metakvCancel:        make(chan struct{}),
		pauseToken:          pauseToken,

		masterTokens:   make(map[string]*PauseUploadToken),
		followerTokens: make(map[string]*PauseUploadToken),
	}

	task.taskMu.Lock()
	task.pauser = pauser
	task.taskMu.Unlock()

	go pauser.observePause()

	if master {
		go pauser.initPauseAsync()
	} else {
		// if not master, no need to wait for publishing of tokens
		close(pauser.waitForTokenPublish)
	}

	// TODO: Move logic from run to handlers for PauseUploadTokens
	go pauser.run(master)
}

////////////////////////////////////////////////////////////////////////////////////////////////////
// Methods
////////////////////////////////////////////////////////////////////////////////////////////////////

func (p *Pauser) initPauseAsync() {

	// TODO: init progress update

	// Generate PauseUploadTokens
	puts, err := p.generatePauseUploadTokens()
	if err != nil {
		logging.Errorf("Pauser::initPauseAsync: Failed to generate PauseUploadTokens: err[%v], puts[%v]",
			err, puts)

		p.finishPause(err)
		return
	}

	// Publish tokens to metaKV
	// will crash if cannot set in metaKV even after retries.
	p.publishPauseUploadTokens(puts)

	// Ask observe to continue
	close(p.waitForTokenPublish)
}

func (p *Pauser) generatePauseUploadTokens() (map[string]*PauseUploadToken, error) {
	indexerUuids, err := p.getIndexerUuids()
	if err != nil || len(indexerUuids) < 1 {
		logging.Errorf("Pauser::generatePauseUploadTokens: Error getting indexer node UUIDs: err[%v]" +
			" indexerUuids[%v]", err, indexerUuids)
		return nil, err
	}

	puts := make(map[string]*PauseUploadToken)
	nodeUUID := string(p.pauseMgr.nodeInfo.NodeID)

	for _, uuid := range indexerUuids {
		rhCb := func(retryAttempt int, lastErr error) error {
			putId, put, err := newPauseUploadToken(nodeUUID, uuid, p.task.taskId, p.task.bucket)
			if err != nil {
				logging.Warnf("Pauser::generatePauseUploadTokens::rhCb: Error making new PauseUploadToken: " +
					"err[%v] retryAttempt[%d] lastErr[%v]", err, retryAttempt, lastErr)
				return err
			}

			if oldPut, ok := puts[putId]; ok {
				err = fmt.Errorf("put with putId[%s] already exists, put[%v]", putId, oldPut)
				logging.Warnf("Pauser::generatePauseUploadTokens::rhCb: PUTId Collision: " +
					"err[%v] retryAttempt[%d] lastErr[%v]", err, retryAttempt, lastErr)
				return err
			}

			puts[putId] = put

			return nil
		}

		rh := common.NewRetryHelper(10, 100 * time.Millisecond, 1, rhCb)
		if err := rh.Run(); err != nil {
			logging.Errorf("Pauser::generatePauseUploadTokens: Failed to generate PauseUploadToken: err[%v]",
				err)
			return nil, err
		}
	}

	return puts, nil
}

func (p *Pauser) getIndexerUuids() (indexerUuids []string, err error) {

	p.pauseMgr.genericMgr.cinfo.Lock()
	defer p.pauseMgr.genericMgr.cinfo.Unlock()

	if err := p.pauseMgr.genericMgr.cinfo.FetchNodesAndSvsInfo(); err != nil {
		logging.Errorf("Pauser::getIndexerUuids Error Fetching Cluster Information %v", err)
		return nil, err
	}

	nids := p.pauseMgr.genericMgr.cinfo.GetNodeIdsByServiceType(common.INDEX_HTTP_SERVICE)
	url := "/nodeuuid"

	for _, nid := range nids {
		haddr, err := p.pauseMgr.genericMgr.cinfo.GetServiceAddress(nid, common.INDEX_HTTP_SERVICE, true)
		if err != nil {
			return nil, err
		}

		resp, err := getWithAuth(haddr + url)
		if err != nil {
			logging.Errorf("Pauser::getIndexerUuids Unable to Fetch Node UUID %v %v", haddr, err)
			return nil, err

		} else {
			defer resp.Body.Close()

			bytes, _ := ioutil.ReadAll(resp.Body)
			uuid := string(bytes)
			indexerUuids = append(indexerUuids, uuid)
		}
	}

	return indexerUuids, nil
}

func (p *Pauser) publishPauseUploadTokens(puts map[string]*PauseUploadToken) {
	for putId, put := range puts {
		setPauseUploadTokenInMetakv(putId, put)
		logging.Infof("Pauser::publishPauseUploadTokens Published pause upload token: %v", putId)
	}
}

func (p *Pauser) observePause() {
	logging.Infof("Pauser::observePause pauseToken[%v] master[%v]", p.pauseToken, p.task.isMaster())

	<-p.waitForTokenPublish

	err := metakv.RunObserveChildren(PauseMetakvDir, p.processUploadTokens, p.metakvCancel)
	if err != nil {
		logging.Errorf("Pauser::observePause Exiting on metaKV observe: err[%v]", err)

		p.finishPause(err)
	}

	logging.Infof("Pauser::observePause exiting: err[%v]", err)
}

// processUploadTokens is metakv callback, not intended to be called otherwise
func (p *Pauser) processUploadTokens(kve metakv.KVEntry) error {

	if kve.Path == buildMetakvPathForPauseToken(p.pauseToken) {
		// Process PauseToken

		logging.Infof("Pauser::processUploadTokens: PauseToken path[%v] value[%s]", kve.Path, kve.Value)

		if kve.Value == nil {
			logging.Infof("Pauser::processUploadTokens: PauseToken Deleted. Mark Done.")
			p.cancelMetakv()
			p.finishPause(nil)
		}

	} else if strings.Contains(kve.Path, PauseUploadTokenPathPrefix) {
		// Process PauseUploadTokens

		if kve.Value != nil {
			putId, put, err := decodePauseUploadToken(kve.Path, kve.Value)
			if err != nil {
				logging.Errorf("Pauser::processUploadTokens: Failed to decode PauseUploadToken. Ignored.")
				return nil
			}

			p.processPauseUploadToken(putId, put)

		} else {
			logging.Infof("Pauser::processUploadTokens: Received empty or deleted PauseUploadToken path[%v]",
				kve.Path)

		}
	}

	return nil
}

func (p *Pauser) cancelMetakv() {
	p.metakvMutex.Lock()
	defer p.metakvMutex.Unlock()

	if p.metakvCancel != nil {
		close(p.metakvCancel)
		p.metakvCancel = nil
	}
}

func (p *Pauser) processPauseUploadToken(putId string, put *PauseUploadToken) {
	logging.Infof("Pauser::processPauseUploadToken putId[%v] put[%v]", putId, put)
	if !p.addToWaitGroup() {
		logging.Errorf("Pauser::processPauseUploadToken: Failed to add to pauser waitgroup.")
		return
	}

	defer p.wg.Done()

	// TODO: Check DDL running

	// "processed" var ensures only the incoming token state gets processed by this
	// call, as metakv will call parent processUploadTokens again for each state change.
	var processed bool

	nodeUUID := string(p.pauseMgr.nodeInfo.NodeID)

	if put.MasterId == nodeUUID {
		processed = p.processPauseUploadTokenAsMaster(putId, put)
	}

	if (put.FollowerId == nodeUUID && !processed) {
		p.processPauseUploadTokenAsFollower(putId, put)
	}
}

func (p *Pauser) addToWaitGroup() bool {
	p.metakvMutex.Lock()
	defer p.metakvMutex.Unlock()

	if p.metakvCancel != nil {
		p.wg.Add(1)
		return true
	}
	return false
}

func (p *Pauser) processPauseUploadTokenAsMaster(putId string, put *PauseUploadToken) bool {

	logging.Infof("Pauser::processPauseUploadTokenAsMaster: putId[%v] put[%v]", putId, put)

	if put.PauseId != p.task.taskId {
		logging.Warnf("Pauser::processPauseUploadTokenAsMaster: Found PauseUploadToken[%v] with Unknown " +
			"PauseId. Expected to match local taskId[%v]", put, p.task.taskId)

		return true
	}

	if put.Error != "" {
		logging.Errorf("Pauser::processPauseUploadTokenAsMaster: Detected PauseUploadToken[%v] in Error state." +
			" Abort.", put)

		p.cancelMetakv()
		go p.finishPause(errors.New(put.Error))

		return true
	}

	if !p.checkValidNotifyState(putId, put, "master") {
		return true
	}

	switch put.State {

	case PauseUploadTokenPosted:
		// Follower owns token, do nothing

		return false

	case PauseUploadTokenInProgess:
		// Follower owns token, just mark in memory maps.

		p.updateInMemToken(putId, put, "master")
		return false

	case PauseUploadTokenProcessed:
		// Master owns token

		// Follower completed work, delete token
		err := common.MetakvDel(PauseMetakvDir + putId)
		if err != nil {
			logging.Fatalf("Pauser::processPauseUploadTokenAsMaster: Failed to delete PauseUploadToken[%v] with" +
				" putId[%v] In Meta Storage: err[%v]", put, putId, err)
			common.CrashOnError(err)
		}

		p.updateInMemToken(putId, put, "master")

		if p.checkAllTokensDone() {
			// All the followers completed work

			// TODO: set progress 100%

			logging.Infof("Pauser::processPauseUploadTokenAsMaster: No Tokens Found. Mark Done.")

			p.cancelMetakv()

			go p.finishPause(nil)
		}

		return true

	default:
		return false

	}

}

func (p *Pauser) finishPause(err error) {

	if p.retErr == nil {
		p.retErr = err
	}

	p.cleanupOnce.Do(p.doFinish)
}

func (p *Pauser) doFinish() {
	logging.Infof("Pauser::doFinish Cleanup: retErr[%v]", p.retErr)

	// TODO: signal others that we are cleaning up using done channel

	p.cancelMetakv()
	p.wg.Wait()

	// TODO: call done callback to start the cleanup phase
}

func (p *Pauser) processPauseUploadTokenAsFollower(putId string, put *PauseUploadToken) bool {

	logging.Infof("Pauser::processPauseUploadTokenAsFollower: putId[%v] put[%v]", putId, put)

	if put.PauseId != p.task.taskId {
		logging.Warnf("Pauser::processPauseUploadTokenAsFollower: Found PauseUploadToken[%v] with Unknown " +
			"PauseId. Expected to match local taskId[%v]", put, p.task.taskId)

		return true
	}

	if !p.checkValidNotifyState(putId, put, "follower") {
		return true
	}

	switch put.State {

	case PauseUploadTokenPosted:
		// Follower owns token, update in-memory and move to InProgress State

		p.updateInMemToken(putId, put, "follower")

		put.State = PauseUploadTokenInProgess
		setPauseUploadTokenInMetakv(putId, put)

		return true

	case PauseUploadTokenInProgess:
		// Follower owns token, update in-memory and start pause work

		p.updateInMemToken(putId, put, "follower")

		go p.startPauseUpload(putId, put)

		return true

	case PauseUploadTokenProcessed:
		// Master owns token, just mark in memory maps

		p.updateInMemToken(putId, put, "follower")

		return false

	default:
		return false
	}
}

func (p *Pauser) startPauseUpload(putId string, put *PauseUploadToken) {
	start := time.Now()
	logging.Infof("Pauser::startPauseUpload: Begin work: putId[%v] put[%v]", putId, put)
	defer logging.Infof("Pauser::startPauseUpload: Done work: putId[%v] put[%v] took[%v]",
		putId, put, time.Since(start))

	if !p.addToWaitGroup() {
		logging.Errorf("Pauser::startPauseUpload: Failed to add to pauser waitgroup.")
		return
	}
	defer p.wg.Done()

	// TODO: Replace sleep with actual work
	time.Sleep(5 * time.Second)

	// work done, change state, master handler will pick it up and do cleanup.
	put.State = PauseUploadTokenProcessed
	setPauseUploadTokenInMetakv(putId, put)
}

// Often, metaKV can send multiple notifications for the same state change (probably
// due to the eventual consistent nature of metaKV). Keep track of all state changes
// in in-memory bookkeeping and ignore the duplicate notifications
func (p *Pauser) checkValidNotifyState(putId string, put *PauseUploadToken, caller string) bool {

	// As the default state is "PauseUploadTokenPosted"
	// do not check for valid state changes for this state
	if put.State == PauseUploadTokenPosted {
		return true
	}

	p.mu.RLock()
	defer p.mu.RUnlock()

	var inMemToken *PauseUploadToken
	var ok bool

	if caller == "master" {
		inMemToken, ok = p.masterTokens[putId]
	} else if caller == "follower" {
		inMemToken, ok = p.followerTokens[putId]
	}

	if ok {
		// Token seen before, validate the state

		// < for invalid state change
		// == for duplicate notification
		if put.State <= inMemToken.State {
			logging.Warnf("Pauser::checkValidNotifyState Detected Invalid State Change Notification" +
				" for [%v]. putId[%v] Local[%v] Metakv[%v]", caller, putId, inMemToken.State, put.State)

			return false
		}
	}

	return true
}

func (p *Pauser) updateInMemToken(putId string, put *PauseUploadToken, caller string) {

	p.mu.Lock()
	defer p.mu.Unlock()

	if caller == "master" {
		p.masterTokens[putId] = put.Clone()
	} else if caller == "follower" {
		p.followerTokens[putId] = put.Clone()
	}
}

func (p *Pauser) checkAllTokensDone() bool {

	p.mu.Lock()
	defer p.mu.Unlock()

	for putId, put := range p.masterTokens {
		if put.State < PauseUploadTokenProcessed {
			// Either posted or processing

			logging.Infof("Pauser::checkAllTokensDone PauseUploadToken: putId[%v] is in state[%v]",
				putId, put.State)

			return false
		}
	}

	return true
}

// restGetLocalIndexMetadataBinary calls the /getLocalndexMetadata REST API (request_handler.go) via
// self-loopback to get the index metadata for the current node and the task's bucket (tenant). This
// verifies it can be unmarshaled, but it returns a checksummed and optionally compressed byte slice
// version of the data rather than the unmarshaled object.
func (this *Pauser) restGetLocalIndexMetadataBinary(compress bool) ([]byte, *manager.LocalIndexMetadata, error) {
	const _restGetLocalIndexMetadataBinary = "Pauser::restGetLocalIndexMetadataBinary:"

	url := fmt.Sprintf("%v/getLocalIndexMetadata?useETag=false&bucket=%v",
		this.pauseMgr.httpAddr, this.task.bucket)
	resp, err := getWithAuth(url)
	if err != nil {
		this.failPause(_restGetLocalIndexMetadataBinary, fmt.Sprintf("getWithAuth(%v)", url), err)
		return nil, nil, err
	}
	defer resp.Body.Close()

	byteSlice, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		this.failPause(_restGetLocalIndexMetadataBinary, "ReadAll(resp.Body)", err)
		return nil, nil, err
	}

	// Verify response can be unmarshaled
	metadata := new(manager.LocalIndexMetadata)
	err = json.Unmarshal(byteSlice, metadata)
	if err != nil {
		this.failPause(_restGetLocalIndexMetadataBinary, "Unmarshal localMeta", err)
		return nil, nil, err
	}
	if len(metadata.IndexDefinitions) == 0 {
		return nil, nil, nil
	}

	// Return checksummed and optionally compressed byte slice, not the unmarshaled object
	return common.ChecksumAndCompress(byteSlice, compress), metadata, nil
}

// failPause logs an error using the caller's logPrefix and a provided context string and aborts the
// Pause task. If there is a set of known Indexer nodes, it will also try to notify them.
func (this *Pauser) failPause(logPrefix string, context string, error error) {
	logging.Errorf("%v Aborting Pause task %v due to %v error: %v", logPrefix,
		this.task.taskId, context, error)

	// Mark the task as failed directly here on master node (avoids dependency on loopback REST)
	this.task.TaskObjSetFailed(error.Error())

	// Notify other Index nodes
	this.pauseMgr.RestNotifyFailedTask(this.otherIndexAddrs, this.task, error.Error())
}

// run is a goroutine for the main body of Pause work for this.task.
//
//	master - true iff this node is the master
func (this *Pauser) run(master bool) {

	// Get the list of Index node host:port addresses EXCLUDING this one
	this.otherIndexAddrs = this.pauseMgr.GetIndexerNodeAddresses(this.pauseMgr.httpAddr)

	var byteSlice []byte
	var err error
	reader := bytes.NewReader(nil)

	dbg := true // TODO: use system config here

	/////////////////////////////////////////////
	// Work done by master only
	/////////////////////////////////////////////

	if master {
		this.task.pauseMetadata.setVersionNoLock(common.GetLocalInternalVersion().String())
		// Write the version.json file to the archive
		// byteSlice = []byte(fmt.Sprintf("{\"version\":%v}\n", ARCHIVE_VERSION))
		// reader.Reset(byteSlice)
		data, err := json.Marshal(this.task.pauseMetadata)
		if err != nil {
			this.failPause("Pauser::run:", "Marshal PauseMetadata", err)
			return
		}
		reader.Reset(common.ChecksumAndCompress(data,!dbg))
		err = Upload(this.task.archivePath, FILENAME_PAUSE_METADATA, reader)
		if err != nil {
			this.failPause("Pauser::run:", "Upload "+FILENAME_PAUSE_METADATA, err)
			return
		}

		logging.Tracef("Pauser::run: indexer version successfully uploaded to %v%v for taskId %v", this.task.archivePath, FILENAME_PAUSE_METADATA, this.task.taskId)

		// Notify the followers to start working on this task
		this.pauseMgr.RestNotifyPause(this.otherIndexAddrs, this.task)
	} // if master

	/////////////////////////////////////////////
	// Work done by both master and followers
	/////////////////////////////////////////////

	// nodePath is the path to the node-specific archive subdirectory for the current node
	nodePath := this.task.archivePath + this.nodeDir

	// Get the index metadata from all nodes and write it as a single file to the archive
	byteSlice, indexMetadata, err := this.restGetLocalIndexMetadataBinary(!dbg)
	if err != nil {
		this.failPause("Pauser::run:", "getLocalInstanceMetadata", err)
		return
	}
	if byteSlice == nil {
		// there are no indexes on this node for bucket. pause is a no-op
		logging.Infof("Pauser::run: pause is a no-op for bucket %v", this.task.bucket)
		return
	}
	reader.Reset(byteSlice)
	err = Upload(nodePath, FILENAME_METADATA, reader)
	if err != nil {
		this.failPause("Pauser::run:", "Upload "+FILENAME_METADATA, err)
		return
	}
	logging.Infof("Pauser::run: metadata successfully uploaded to %v%v for taskId %v", nodePath, FILENAME_METADATA, this.task.taskId)

	getIndexInstanceIds := func() []common.IndexInstId {
		res := make([]common.IndexInstId, 0, len(indexMetadata.IndexDefinitions))
		for _, topology := range indexMetadata.IndexTopologies {
			for _, indexDefn := range topology.Definitions {
				res = append(res, common.IndexInstId(indexDefn.Instances[0].InstId))
			}
		}
		logging.Tracef("Pauser::run::getIndexInstanceId: index instance ids: %v for bucket %v", res, this.task.bucket)
		return res
	}

	// Write the persistent stats to the archive
	byteSlice, err = this.pauseMgr.genericMgr.statsMgr.GetStatsForIndexesToBePersisted(getIndexInstanceIds(), !dbg)
	if err != nil {
		this.failPause("Pauser::run:", "GetStatsForIndexesToBePersisted", err)
		return
	}
	reader.Reset(byteSlice)
	err = Upload(nodePath, FILENAME_STATS, reader)
	if err != nil {
		this.failPause("Pauser::run:", "Upload "+FILENAME_STATS, err)
		return
	}

	logging.Infof("Pauser::run: stats successfully uploaded to %v%v for taskId %v", nodePath, FILENAME_STATS, this.task.taskId)

	getShardIds := func() []common.ShardId {
		uniqueShardIds := make(map[common.ShardId]bool)
		for _, topology := range indexMetadata.IndexTopologies {
			for _, indexDefn := range topology.Definitions {
				for _, instance := range indexDefn.Instances {
					for _, partition := range instance.Partitions {
						for _, shard := range partition.ShardIds {
							uniqueShardIds[shard] = true
						}
					}

				}
			}
		}

		shardIds := make([]common.ShardId, 0, len(uniqueShardIds))
		for shardId, _ := range uniqueShardIds {
			shardIds = append(shardIds, shardId)
		}
		logging.Tracef("Pauser::run::getShardIds: found shard Ids %v for bucket %v", shardIds, this.task.bucket)

		return shardIds
	}

	// TODO: add contextWithCancel to task and reuse it here
	ctx := context.Background()
	closeCh := ctx.Done()
	respCh := make(chan Message)
	// progressCh := make(chan *ShardTransferStatistics, 1000) TODO: progress reporting

	msg := &MsgStartShardTransfer{
		shardIds:    getShardIds(),
		taskId:      this.task.taskId,
		transferId:  this.task.bucket,
		taskType:    common.PauseResumeTask,
		destination: nodePath,

		cancelCh:   closeCh,
		doneCh:     closeCh,
		respCh:     respCh,
		progressCh: nil,
	}

	this.pauseMgr.supvMsgch <- msg

	resp, ok := (<-respCh).(*MsgShardTransferResp)
	if !ok || resp == nil {
		err := fmt.Errorf("couldn't get a valid response from ShardTransferManager")
		this.failPause("Pauser::run", "Upload plasma shards", err)
		logging.Errorf("Pauser::run %v for taskId %v", err, this.task.taskId)
		return
	}
	errMap := resp.GetErrorMap()
	var errMsg strings.Builder
	for shard, err := range errMap {
		if err != nil {
			fmt.Fprintf(&errMsg, "Error in shardId %v upload: %v\n", shard, err)
		}
	}
	if errMsg.Len() != 0 {
		err = errors.New(errMsg.String())
		this.failPause("Pauser::run", "Upload plasma shards", err)
		logging.Errorf("Pauser::run shard uploads failed: %v for taskId %v", err, this.task.taskId)
		return
	}

	// TODO: return shard paths back to master
	logging.Infof("Pauser::run Shards saved at: %v for bucket %v", resp.GetShardPaths(), this.task.bucket)
}