xaction.go

// Package api provides AIStore API over HTTP(S)
/*
 * Copyright (c) 2018-2021, NVIDIA CORPORATION. All rights reserved.
 */
package api

import (
	"context"
	"fmt"
	"net/http"
	"net/url"
	"time"

	"github.com/NVIDIA/aistore/api/apc"
	"github.com/NVIDIA/aistore/cmn"
	"github.com/NVIDIA/aistore/cmn/cos"
	"github.com/NVIDIA/aistore/cmn/debug"
	"github.com/NVIDIA/aistore/nl"
	"github.com/NVIDIA/aistore/xact"
)

const (
	XactDefWaitTimeShort = time.Minute
	XactDefWaitTimeLong  = 7 * 24 * time.Hour // week
	XactMaxStartPollTime = time.Minute
	XactMaxPollTime      = time.Hour
	XactPollTime         = time.Second
	numConsecutiveIdle   = 3 // number of consecutive 'idle' states
)

type (
	NodesXactSnap      map[string]*xact.SnapExt
	NodesXactMultiSnap map[string][]*xact.SnapExt
	XactSnapTestFunc   func(NodesXactMultiSnap) bool

	XactStatsHelper interface {
		Running() bool
		Finished() bool
		Aborted() bool
		ObjCount() int64
	}

	XactReqArgs struct {
		// either xaction ID or Kind _must_ be specified
		ID   string // xaction UUID
		Kind string // xaction kind, see `xact.Table`)
		// optional parameters to further narrow down or filter out xactions in question
		DaemonID string    // node that runs this xaction
		Bck      cmn.Bck   // bucket
		Buckets  []cmn.Bck // list of buckets (e.g., copy-bucket, lru-evict, etc.)
		// max time to wait and other "non-filters"
		Timeout time.Duration
		Force   bool // force
		// more filters
		OnlyRunning bool // look only for running xactions
	}
)

// StartXaction starts a given xact.
func StartXaction(baseParams BaseParams, args XactReqArgs) (id string, err error) {
	if !xact.Table[args.Kind].Startable {
		return id, fmt.Errorf("cannot start \"kind=%s\" xaction", args.Kind)
	}
	xactMsg := xact.QueryMsg{Kind: args.Kind, Bck: args.Bck, DaemonID: args.DaemonID}
	if args.Kind == apc.ActLRU {
		ext := &xact.QueryMsgLRU{}
		if args.Buckets != nil {
			xactMsg.Buckets = args.Buckets
			ext.Force = args.Force
		}
		xactMsg.Ext = ext
	} else if args.Kind == apc.ActStoreCleanup && args.Buckets != nil {
		xactMsg.Buckets = args.Buckets
	}

	msg := apc.ActionMsg{Action: apc.ActXactStart, Value: xactMsg}
	baseParams.Method = http.MethodPut
	reqParams := allocRp()
	{
		reqParams.BaseParams = baseParams
		reqParams.Path = apc.URLPathClu.S
		reqParams.Body = cos.MustMarshal(msg)
		reqParams.Header = http.Header{cmn.HdrContentType: []string{cmn.ContentJSON}}
		reqParams.Query = args.Bck.AddToQuery(nil)
	}
	err = reqParams.DoHTTPReqResp(&id)
	freeRp(reqParams)
	return id, err
}

// AbortXaction aborts a given xact.
func AbortXaction(baseParams BaseParams, args XactReqArgs) error {
	msg := apc.ActionMsg{
		Action: apc.ActXactStop,
		Value:  xact.QueryMsg{ID: args.ID, Kind: args.Kind, Bck: args.Bck},
	}
	baseParams.Method = http.MethodPut
	reqParams := allocRp()
	{
		reqParams.BaseParams = baseParams
		reqParams.Path = apc.URLPathClu.S
		reqParams.Body = cos.MustMarshal(msg)
		reqParams.Header = http.Header{cmn.HdrContentType: []string{cmn.ContentJSON}}
		reqParams.Query = args.Bck.AddToQuery(nil)
	}
	err := reqParams.DoHTTPRequest()
	freeRp(reqParams)
	return err
}

// GetXactionSnapsByID gets all xaction snaps for a given xaction id.
func GetXactionSnapsByID(baseParams BaseParams, xactID string) (nxs NodesXactSnap, err error) {
	xs, err := QueryXactionSnaps(baseParams, XactReqArgs{ID: xactID})
	if err != nil {
		return
	}
	nxs = xs.nodesXactSnap(xactID)
	return
}

// QueryXactionSnaps gets all xaction snaps based on the specified selection.
func QueryXactionSnaps(baseParams BaseParams, args XactReqArgs) (xs NodesXactMultiSnap, err error) {
	msg := xact.QueryMsg{ID: args.ID, Kind: args.Kind, Bck: args.Bck}
	if args.OnlyRunning {
		msg.OnlyRunning = Bool(true)
	}
	baseParams.Method = http.MethodGet
	reqParams := allocRp()
	{
		reqParams.BaseParams = baseParams
		reqParams.Path = apc.URLPathClu.S
		reqParams.Body = cos.MustMarshal(msg)
		reqParams.Header = http.Header{cmn.HdrContentType: []string{cmn.ContentJSON}}
		reqParams.Query = url.Values{apc.QparamWhat: []string{apc.GetWhatQueryXactStats}}
	}
	err = reqParams.DoHTTPReqResp(&xs)
	freeRp(reqParams)
	return xs, err
}

// GetXactionStatus retrieves the status of the xact.
func GetXactionStatus(baseParams BaseParams, args XactReqArgs) (status *nl.NotifStatus, err error) {
	baseParams.Method = http.MethodGet
	msg := xact.QueryMsg{ID: args.ID, Kind: args.Kind, Bck: args.Bck}
	if args.OnlyRunning {
		msg.OnlyRunning = Bool(true)
	}
	status = &nl.NotifStatus{}
	reqParams := allocRp()
	{
		reqParams.BaseParams = baseParams
		reqParams.Path = apc.URLPathClu.S
		reqParams.Body = cos.MustMarshal(msg)
		reqParams.Header = http.Header{cmn.HdrContentType: []string{cmn.ContentJSON}}
		reqParams.Query = url.Values{apc.QparamWhat: []string{apc.GetWhatStatus}}
	}
	err = reqParams.DoHTTPReqResp(status)
	freeRp(reqParams)
	return
}

func initPollingTimes(args XactReqArgs) (time.Duration, time.Duration) {
	total := args.Timeout
	switch {
	case args.Timeout == 0:
		total = XactDefWaitTimeShort
	case args.Timeout < 0:
		total = XactDefWaitTimeLong
	}
	return total, cos.MinDuration(XactMaxStartPollTime, cos.ProbingFrequency(total))
}

func backoffPoll(dur time.Duration) time.Duration {
	dur += dur / 2
	return cos.MinDuration(XactMaxPollTime, dur)
}

func _waitForXaction(baseParams BaseParams, args XactReqArgs, condFn ...XactSnapTestFunc) (status *nl.NotifStatus, err error) {
	var (
		total, sleep = initPollingTimes(args)
		ctx, cancel  = context.WithTimeout(context.Background(), total)
	)
	defer cancel()
	for {
		var done bool
		if len(condFn) == 0 {
			status, err = GetXactionStatus(baseParams, args)
			done = err == nil && status.Finished()
		} else {
			var (
				snaps NodesXactMultiSnap
				fn    = condFn[0]
			)
			snaps, err = QueryXactionSnaps(baseParams, args)
			done = err == nil && fn(snaps)
		}
		canRetry := err == nil || cos.IsRetriableConnErr(err) || cmn.IsStatusServiceUnavailable(err)
		if done || !canRetry /*fail*/ {
			return
		}
		time.Sleep(sleep)
		sleep = backoffPoll(sleep)
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		default:
			break
		}
	}
}

// WaitForXactionIC waits for a given xaction to complete.
// Use it only for global xactions(e.g rebalance) that reports their statuses to IC.
func WaitForXactionIC(baseParams BaseParams, args XactReqArgs) (status *nl.NotifStatus, err error) {
	return _waitForXaction(baseParams, args)
}

// WaitForXactionNode waits for a given xaction to complete.
// Use for xaction which can be launched on a single node and do not report their
// statuses(e.g resilver) to IC or to check specific xaction states (e.g Idle).
func WaitForXactionNode(baseParams BaseParams, args XactReqArgs, fn XactSnapTestFunc) error {
	debug.Assert(args.Kind != "")
	_, err := _waitForXaction(baseParams, args, fn)
	return err
}

// WaitForXactionIdle waits for a given on-demand xaction to be idle.
func WaitForXactionIdle(baseParams BaseParams, args XactReqArgs) error {
	idles := 0
	args.OnlyRunning = true
	check := func(snaps NodesXactMultiSnap) bool {
		if snaps.Idle() {
			idles++
			return idles >= numConsecutiveIdle
		}
		idles = 0
		return false
	}
	return WaitForXactionNode(baseParams, args, check)
}

///////////////////
// NodesXactSnap //
///////////////////

func (nxs NodesXactSnap) running() bool {
	for _, snap := range nxs {
		if snap.Running() {
			return true
		}
	}
	return false
}

func (nxs NodesXactSnap) Finished() bool { return !nxs.running() }

func (nxs NodesXactSnap) IsAborted() bool {
	for _, snap := range nxs {
		if snap.IsAborted() {
			return true
		}
	}
	return false
}

func (nxs NodesXactSnap) ObjCounts() (locObjs, outObjs, inObjs int64) {
	for _, snap := range nxs {
		locObjs += snap.Stats.Objs
		outObjs += snap.Stats.OutObjs
		inObjs += snap.Stats.InObjs
	}
	return
}

func (nxs NodesXactSnap) ByteCounts() (locBytes, outBytes, inBytes int64) {
	for _, snap := range nxs {
		locBytes += snap.Stats.Bytes
		outBytes += snap.Stats.OutBytes
		inBytes += snap.Stats.InBytes
	}
	return
}

func (nxs NodesXactSnap) TotalRunningTime() time.Duration {
	var (
		start   = time.Now()
		end     time.Time
		running bool
	)
	for _, snap := range nxs {
		running = running || snap.Running()
		if snap.StartTime.Before(start) {
			start = snap.StartTime
		}
		if snap.EndTime.After(end) {
			end = snap.EndTime
		}
	}
	if running {
		end = time.Now()
	}
	return end.Sub(start)
}

////////////////////////
// NodesXactMultiSnap //
////////////////////////

func (xs NodesXactMultiSnap) Running() (tid string, xsnap *xact.SnapExt) {
	var snaps []*xact.SnapExt
	for tid, snaps = range xs {
		for _, xsnap = range snaps {
			if xsnap.Running() {
				return
			}
		}
	}
	return "", nil
}

func (xs NodesXactMultiSnap) ObjCounts() (locObjs, outObjs, inObjs int64) {
	for _, targetStats := range xs {
		for _, snap := range targetStats {
			locObjs += snap.Stats.Objs
			outObjs += snap.Stats.OutObjs
			inObjs += snap.Stats.InObjs
		}
	}
	return
}

func (xs NodesXactMultiSnap) ByteCounts() (locBytes, outBytes, inBytes int64) {
	for _, targetStats := range xs {
		for _, snap := range targetStats {
			locBytes += snap.Stats.Bytes
			outBytes += snap.Stats.OutBytes
			inBytes += snap.Stats.InBytes
		}
	}
	return
}

func (xs NodesXactMultiSnap) nodesXactSnap(xactID string) (nxs NodesXactSnap) {
	nxs = make(NodesXactSnap)
	for tid, snaps := range xs {
		for _, xsnap := range snaps {
			if xsnap.ID == xactID {
				nxs[tid] = xsnap
				break
			}
		}
	}
	return
}

func (xs NodesXactMultiSnap) Idle() bool {
	for _, xs := range xs {
		for _, xsnap := range xs {
			if !xsnap.Idle() {
				return false
			}
		}
	}
	return true
}