monitor.go

package lib

import (
	"fmt"
	"strings"
	"sync/atomic"
	"time"
)

const (
	normalExit = iota
	errExit
)

var processTickInterval int64 = 5

var clearStrLen int = 0
var clearStr string = strings.Repeat(" ", clearStrLen)

func getClearStr(str string) string {
	if clearStrLen <= len(str) {
		clearStrLen = len(str)
		return fmt.Sprintf("\r%s", str)
	}
	clearStr = strings.Repeat(" ", clearStrLen)
	return fmt.Sprintf("\r%s\r%s", clearStr, str)
}

type Monitorer interface {
	setScanError(err error)
	updateScanNum(num int64)
	setScanEnd()
}

// for normal object operation
type MonitorSnap struct {
	okNum   int64
	errNum  int64
	dealNum int64
}

/*
 * Put same type variables together to make them 64bits alignment to avoid
 * atomic.AddInt64() panic
 * Please guarantee the alignment if you add new filed
 */
type Monitor struct {
	opStr          string
	totalNum       int64
	okNum          int64
	errNum         int64
	seekAheadError error
	seekAheadEnd   bool
	finish         bool
	_              uint32 //Add padding to make sure the next data 64bits alignment
}

func (m *Monitor) init(opStr string) {
	m.opStr = opStr
	m.totalNum = 0
	m.seekAheadEnd = false
	m.seekAheadError = nil
	m.okNum = 0
	m.errNum = 0
	m.finish = false
}

func (m *Monitor) setScanError(err error) {
	m.seekAheadError = err
	m.seekAheadEnd = true
}

func (m *Monitor) updateScanNum(num int64) {
	m.totalNum = m.totalNum + num
}

func (m *Monitor) setScanEnd() {
	m.seekAheadEnd = true
}

func (m *Monitor) updateOKNum(num int64) {
	atomic.AddInt64(&m.okNum, num)
}

func (m *Monitor) updateErrNum(num int64) {
	atomic.AddInt64(&m.errNum, num)
}

func (m *Monitor) getSnapshot() *MonitorSnap {
	var snap MonitorSnap
	snap.okNum = m.okNum
	snap.errNum = m.errNum
	snap.dealNum = snap.okNum + snap.errNum
	return &snap
}

func (m *Monitor) progressBar(finish bool, exitStat int) string {
	if m.finish {
		return ""
	}
	m.finish = m.finish || finish
	if !finish {
		return m.getProgressBar()
	}
	return m.getFinishBar(exitStat)
}

func (m *Monitor) getProgressBar() string {
	snap := m.getSnapshot()
	if m.seekAheadEnd && m.seekAheadError == nil {
		if snap.errNum == 0 {
			return getClearStr(fmt.Sprintf("Total %d objects. %s %d objects, Progress: %d%s", m.totalNum, m.opStr, snap.okNum, m.getPrecent(snap), "%%"))
		}
		return getClearStr(fmt.Sprintf("Total %d objects. %s %d objects, Error %d objects, Progress: %d%s", m.totalNum, m.opStr, snap.okNum, snap.errNum, m.getPrecent(snap), "%%"))
	}
	scanNum := max(m.totalNum, snap.dealNum)
	if snap.errNum == 0 {
		return getClearStr(fmt.Sprintf("Scanned %d objects. %s %d objects.", scanNum, m.opStr, snap.okNum))
	}
	return getClearStr(fmt.Sprintf("Scanned %d objects. %s %d objects, Error %d objects.", scanNum, m.opStr, snap.okNum, snap.errNum))
}

func (m *Monitor) getPrecent(snap *MonitorSnap) int {
	if m.seekAheadEnd && m.seekAheadError == nil {
		if m.totalNum != 0 {
			return int(float64((snap.dealNum)*100.0) / float64(m.totalNum))
		}
		return 100
	}
	return 0
}

func (m *Monitor) getFinishBar(exitStat int) string {
	if exitStat == normalExit {
		return m.getWholeFinishBar()
	}
	return m.getDefeatBar()
}

func (m *Monitor) getWholeFinishBar() string {
	snap := m.getSnapshot()
	if m.seekAheadEnd && m.seekAheadError == nil {
		if snap.errNum == 0 {
			return getClearStr(fmt.Sprintf("Succeed: Total %d objects. %s %d objects.\n", m.totalNum, m.opStr, snap.okNum))
		}
		return getClearStr(fmt.Sprintf("FinishWithError: Total %d objects. %s %d objects, Error %d objects.\n", m.totalNum, m.opStr, snap.okNum, snap.errNum))
	}
	scanNum := max(m.totalNum, snap.dealNum)
	if snap.errNum == 0 {
		return getClearStr(fmt.Sprintf("Succeed: Total %d objects. %s %d objects.\n", scanNum, m.opStr, snap.okNum))
	}
	return getClearStr(fmt.Sprintf("FinishWithError: Scanned %d objects. %s %d objects, Error %d objects.\n", scanNum, m.opStr, snap.okNum, snap.errNum))
}

func (m *Monitor) getDefeatBar() string {
	snap := m.getSnapshot()
	if m.seekAheadEnd && m.seekAheadError == nil {
		return getClearStr(fmt.Sprintf("Total %d objects. %s %d objects, when error happens.\n", m.totalNum, m.opStr, snap.okNum))
	}
	scanNum := max(m.totalNum, snap.dealNum)
	return getClearStr(fmt.Sprintf("Scanned %d objects. %s %d objects, when error happens.\n", scanNum, m.opStr, snap.okNum))
}

// For rm
type RMMonitorSnap struct {
	objectNum      int64
	uploadIdNum    int64
	errObjectNum   int64
	errUploadIdNum int64
	dealNum        int64
	errNum         int64
	removedBucket  string
}

/*
 * Put same type variables together to make them 64bits alignment to avoid
 * atomic.AddInt64() panic
 * Please guarantee the alignment if you add new filed
 */
type RMMonitor struct {
	op               int64
	totalObjectNum   int64
	totalUploadIdNum int64
	objectNum        int64
	uploadIdNum      int64
	errObjectNum     int64
	errUploadIdNum   int64
	removedBucket    string
	seekAheadError   error
	seekAheadEnd     bool
	finish           bool
	_                uint32 //Add padding to make sure the next data 64bits alignment
}

func (m *RMMonitor) init() {
	m.op = 0
	m.totalObjectNum = 0
	m.totalUploadIdNum = 0
	m.seekAheadEnd = false
	m.seekAheadError = nil
	m.objectNum = 0
	m.uploadIdNum = 0
	m.errObjectNum = 0
	m.errUploadIdNum = 0
	m.finish = false
	m.removedBucket = ""
}

func (m *RMMonitor) updateOP(op int64) {
	m.op = m.op | op
}

func (m *RMMonitor) setOP(op int64) {
	m.op = op
}

func (m *RMMonitor) setScanError(err error) {
	m.seekAheadError = err
	m.seekAheadEnd = true
}

func (m *RMMonitor) updateScanNum(num int64) {
	m.totalObjectNum = m.totalObjectNum + num
}

func (m *RMMonitor) updateScanUploadIdNum(num int64) {
	m.totalUploadIdNum = m.totalUploadIdNum + num
}

func (m *RMMonitor) setScanEnd() {
	m.seekAheadEnd = true
}

func (m *RMMonitor) updateObjectNum(num int64) {
	atomic.AddInt64(&m.objectNum, num)
}

func (m *RMMonitor) updateUploadIdNum(num int64) {
	atomic.AddInt64(&m.uploadIdNum, num)
}

func (m *RMMonitor) updateErrObjectNum(num int64) {
	atomic.AddInt64(&m.errObjectNum, num)
}

func (m *RMMonitor) updateErrUploadIdNum(num int64) {
	atomic.AddInt64(&m.errUploadIdNum, num)
}

func (m *RMMonitor) updateRemovedBucket(bucket string) {
	m.removedBucket = bucket
}

func (m *RMMonitor) getSnapshot() *RMMonitorSnap {
	var snap RMMonitorSnap
	snap.objectNum = m.objectNum
	snap.uploadIdNum = m.uploadIdNum
	snap.errObjectNum = m.errObjectNum
	snap.errUploadIdNum = m.errUploadIdNum
	snap.dealNum = snap.objectNum + snap.uploadIdNum + snap.errObjectNum + snap.errUploadIdNum
	snap.errNum = snap.errObjectNum + snap.errUploadIdNum
	snap.removedBucket = m.removedBucket
	return &snap
}

func (m *RMMonitor) progressBar(finish bool, exitStat int) string {
	if m.finish {
		return ""
	}
	m.finish = m.finish || finish
	if !finish {
		return m.getProgressBar()
	}
	return m.getFinishBar(exitStat)
}

func (m *RMMonitor) getProgressBar() string {
	if m.op&allType != 0 {
		snap := m.getSnapshot()
		if m.seekAheadEnd && m.seekAheadError == nil {
			return getClearStr(fmt.Sprintf("Total %s. %s%s Progress: %d%s", m.getTotalInfo(), m.getOKInfo(snap), m.getErrInfo(snap), m.getPrecent(snap), "%%"))
		}
		m.totalObjectNum = max(m.totalObjectNum, snap.objectNum+snap.errObjectNum)
		m.totalUploadIdNum = max(m.totalUploadIdNum, snap.uploadIdNum+snap.errUploadIdNum)
		return getClearStr(fmt.Sprintf("Scanned %s. %s%s", m.getTotalInfo(), m.getOKInfo(snap), m.getErrInfo(snap)))
	}
	return getClearStr("")
}

func (m *RMMonitor) getTotalInfo() string {
	strList := []string{}
	if m.op&objectType != 0 {
		strList = append(strList, fmt.Sprintf("%d objects", m.totalObjectNum))
	}
	if m.op&multipartType != 0 {
		strList = append(strList, fmt.Sprintf("%d uploadIds", m.totalUploadIdNum))
	}
	return strings.Join(strList, ", ")
}

func (m *RMMonitor) getOKInfo(snap *RMMonitorSnap) string {
	strList := []string{}
	if m.op&allType == 0 {
		return ""
	}
	if m.op&objectType != 0 {
		strList = append(strList, fmt.Sprintf("%d objects", snap.objectNum))
	}
	if m.op&multipartType != 0 {
		strList = append(strList, fmt.Sprintf("%d uploadIds", snap.uploadIdNum))
	}
	return fmt.Sprintf("Removed %s.", strings.Join(strList, ", "))
}

func (m *RMMonitor) getErrInfo(snap *RMMonitorSnap) string {
	if snap.errNum != 0 {
		strList := []string{}
		if snap.errObjectNum != 0 {
			strList = append(strList, fmt.Sprintf("%d objects", snap.errObjectNum))
		}
		if snap.errUploadIdNum != 0 {
			strList = append(strList, fmt.Sprintf("%d uploadIds", snap.errUploadIdNum))
		}
		return fmt.Sprintf(" Error %s.", strings.Join(strList, ", "))
	}
	return ""
}

func (m *RMMonitor) getPrecent(snap *RMMonitorSnap) int {
	if m.seekAheadEnd && m.seekAheadError == nil {
		if m.totalObjectNum+m.totalUploadIdNum != 0 {
			return int(float64((snap.dealNum)*100.0) / float64(m.totalObjectNum+m.totalUploadIdNum))
		}
		return 100
	}
	return 0
}

func (m *RMMonitor) getFinishBar(exitStat int) string {
	snap := m.getSnapshot()
	return m.getObjectFinishBar(snap, exitStat) + m.getBucketFinishBar(snap)
}

func (m *RMMonitor) getObjectFinishBar(snap *RMMonitorSnap, exitStat int) string {
	if m.op&allType != 0 {
		if m.seekAheadEnd && m.seekAheadError == nil {
			if m.getExitStat(snap, exitStat) == errExit {
				return getClearStr(fmt.Sprintf("Total %s. %s when error happens.\n", m.getTotalInfo(), m.getOKInfo(snap)))
			}
			return getClearStr(fmt.Sprintf("Succeed: Total %s. %s\n", m.getTotalInfo(), m.getOKInfo(snap)))
		}
		m.totalObjectNum = max(m.totalObjectNum, snap.objectNum+snap.errObjectNum)
		m.totalUploadIdNum = max(m.totalUploadIdNum, snap.uploadIdNum+snap.errUploadIdNum)
		if m.getExitStat(snap, exitStat) == errExit {
			return getClearStr(fmt.Sprintf("Scanned %s. %s when error happens.\n", m.getTotalInfo(), m.getOKInfo(snap)))
		}
		return getClearStr(fmt.Sprintf("Succeed: Total %s. %s\n", m.getTotalInfo(), m.getOKInfo(snap)))
	}
	return getClearStr("")
}

func (m *RMMonitor) getExitStat(snap *RMMonitorSnap, exitStat int) int {
	if exitStat != normalExit || snap.errNum != 0 || (m.op&bucketType != 0 && snap.removedBucket == "") {
		return errExit
	}
	return normalExit
}

func (m *RMMonitor) getBucketFinishBar(snap *RMMonitorSnap) string {
	if m.op&bucketType != 0 && snap.removedBucket != "" {
		return getClearStr(fmt.Sprintf("Removed Bucket: %s\n", snap.removedBucket))
	}
	return getClearStr("")
}

// for cp
type CPMonitorSnap struct {
	transferSize  int64
	skipSize      int64
	dealSize      int64
	fileNum       int64
	dirNum        int64
	skipNum       int64
	errNum        int64
	okNum         int64
	dealNum       int64
	duration      int64
	incrementSize int64
}

/*
 * Put same type variables together to make them 64bits alignment to avoid
 * atomic.AddInt64() panic
 * Please guarantee the alignment if you add new filed
 */
type CPMonitor struct {
	totalSize      int64
	totalNum       int64
	transferSize   int64
	skipSize       int64
	fileNum        int64
	dirNum         int64
	skipNum        int64
	errNum         int64
	lastSnapSize   int64
	tickDuration   int64
	seekAheadError error
	op             operationType
	seekAheadEnd   bool
	finish         bool
	_              uint32 //Add padding to make sure the next data 64bits alignment
	lastSnapTime   time.Time
}

func (m *CPMonitor) init(op operationType) {
	m.op = op
	m.totalSize = 0
	m.totalNum = 0
	m.seekAheadEnd = false
	m.seekAheadError = nil
	m.transferSize = 0
	m.skipSize = 0
	m.fileNum = 0
	m.dirNum = 0
	m.skipNum = 0
	m.errNum = 0
	m.finish = false
	m.lastSnapSize = 0
	m.lastSnapTime = time.Now()
	m.tickDuration = processTickInterval * int64(time.Second)
}

func (m *CPMonitor) setScanError(err error) {
	m.seekAheadError = err
	m.seekAheadEnd = true
}

func (m *CPMonitor) updateScanNum(num int64) {
	m.totalNum = m.totalNum + num
}

func (m *CPMonitor) updateScanSizeNum(size, num int64) {
	m.totalSize = m.totalSize + size
	m.totalNum = m.totalNum + num
}

func (m *CPMonitor) setScanEnd() {
	m.seekAheadEnd = true
}

func (m *CPMonitor) updateTransferSize(size int64) {
	atomic.AddInt64(&m.transferSize, size)
}

func (m *CPMonitor) updateFile(size, num int64) {
	atomic.AddInt64(&m.fileNum, num)
	atomic.AddInt64(&m.transferSize, size)
}

func (m *CPMonitor) updateDir(size, num int64) {
	atomic.AddInt64(&m.dirNum, num)
	atomic.AddInt64(&m.transferSize, size)
}

func (m *CPMonitor) updateSkip(size, num int64) {
	atomic.AddInt64(&m.skipNum, num)
	atomic.AddInt64(&m.skipSize, size)
}

func (m *CPMonitor) updateErr(size, num int64) {
	atomic.AddInt64(&m.errNum, num)
	atomic.AddInt64(&m.transferSize, size)
}

func (m *CPMonitor) getSnapshot() *CPMonitorSnap {
	var snap CPMonitorSnap
	snap.transferSize = m.transferSize
	snap.skipSize = m.skipSize
	snap.dealSize = snap.transferSize + snap.skipSize
	snap.fileNum = m.fileNum
	snap.dirNum = m.dirNum
	snap.skipNum = m.skipNum
	snap.errNum = m.errNum
	snap.okNum = snap.fileNum + snap.dirNum + snap.skipNum
	snap.dealNum = snap.okNum + snap.errNum
	now := time.Now()
	snap.duration = now.Sub(m.lastSnapTime).Nanoseconds()

	return &snap
}

func (m *CPMonitor) progressBar(finish bool, exitStat int) string {
	if m.finish {
		return ""
	}
	m.finish = m.finish || finish
	if !finish {
		return m.getProgressBar()
	}
	return m.getFinishBar(exitStat)
}

func (m *CPMonitor) getProgressBar() string {
	mu.RLock()
	defer mu.RUnlock()

	snap := m.getSnapshot()
	if snap.duration < m.tickDuration {
		return ""
	} else {
		m.lastSnapTime = time.Now()
		snap.incrementSize = m.transferSize - m.lastSnapSize
		m.lastSnapSize = snap.transferSize
	}

	if m.seekAheadEnd && m.seekAheadError == nil {
		return getClearStr(fmt.Sprintf("Total num: %d, size: %s. Dealed num: %d%s%s, Progress: %.3f%s, Speed: %.2fKB/s", m.totalNum, getSizeString(m.totalSize), snap.dealNum, m.getDealNumDetail(snap), m.getDealSizeDetail(snap), m.getPrecent(snap), "%%", m.getSpeed(snap)))
	}
	scanNum := max(m.totalNum, snap.dealNum)
	scanSize := max(m.totalSize, snap.dealSize)
	return getClearStr(fmt.Sprintf("Scanned num: %d, size: %s. Dealed num: %d%s%s, Speed: %.2fKB/s.", scanNum, getSizeString(scanSize), snap.dealNum, m.getDealNumDetail(snap), m.getDealSizeDetail(snap), m.getSpeed(snap)))
}

func (m *CPMonitor) getFinishBar(exitStat int) string {
	if exitStat == normalExit {
		return m.getWholeFinishBar()
	}
	return m.getDefeatBar()
}

func (m *CPMonitor) getWholeFinishBar() string {
	snap := m.getSnapshot()
	if m.seekAheadEnd && m.seekAheadError == nil {
		if snap.errNum == 0 {
			return getClearStr(fmt.Sprintf("Succeed: Total num: %d, size: %s. OK num: %d%s%s.\n", m.totalNum, getSizeString(m.totalSize), snap.okNum, m.getDealNumDetail(snap), m.getSkipSize(snap)))
		}
		return getClearStr(fmt.Sprintf("FinishWithError: Total num: %d, size: %s. Error num: %d. OK num: %d%s%s.\n", m.totalNum, getSizeString(m.totalSize), snap.errNum, snap.okNum, m.getOKNumDetail(snap), m.getSizeDetail(snap)))
	}
	scanNum := max(m.totalNum, snap.dealNum)
	if snap.errNum == 0 {
		return getClearStr(fmt.Sprintf("Succeed: Total num: %d, size: %s. OK num: %d%s%s.\n", scanNum, getSizeString(snap.dealSize), snap.okNum, m.getDealNumDetail(snap), m.getSkipSize(snap)))
	}
	return getClearStr(fmt.Sprintf("FinishWithError: Scanned %d %s. Error num: %d. OK num: %d%s%s.\n", scanNum, m.getSubject(), snap.errNum, snap.okNum, m.getOKNumDetail(snap), m.getSizeDetail(snap)))
}

func (m *CPMonitor) getDefeatBar() string {
	snap := m.getSnapshot()
	if m.seekAheadEnd && m.seekAheadError == nil {
		return getClearStr(fmt.Sprintf("Total num: %d, size: %s. Dealed num: %d%s%s. When error happens.\n", m.totalNum, getSizeString(m.totalSize), snap.okNum, m.getOKNumDetail(snap), m.getSizeDetail(snap)))
	}
	scanNum := max(m.totalNum, snap.dealNum)
	return getClearStr(fmt.Sprintf("Scanned %d %s. Dealed num: %d%s%s. When error happens.\n", scanNum, m.getSubject(), snap.okNum, m.getOKNumDetail(snap), m.getSizeDetail(snap)))
}

func (m *CPMonitor) getSubject() string {
	switch m.op {
	case operationTypePut:
		return "files"
	default:
		return "objects"
	}
}

func (m *CPMonitor) getDealNumDetail(snap *CPMonitorSnap) string {
	return m.getNumDetail(snap, true)
}

func (m *CPMonitor) getOKNumDetail(snap *CPMonitorSnap) string {
	return m.getNumDetail(snap, false)
}

func (m *CPMonitor) getNumDetail(snap *CPMonitorSnap, hasErr bool) string {
	if snap.dealNum == 0 || (!hasErr && snap.okNum == 0) {
		return ""
	}
	strList := []string{}
	if hasErr && snap.errNum != 0 {
		strList = append(strList, fmt.Sprintf("Error %d %s", snap.errNum, m.getSubject()))
	}
	if snap.fileNum != 0 {
		strList = append(strList, fmt.Sprintf("%s %d %s", m.getOPStr(), snap.fileNum, m.getSubject()))
	}
	if snap.dirNum != 0 {
		str := fmt.Sprintf("%d directories", snap.dirNum)
		if snap.fileNum == 0 {
			str = fmt.Sprintf("%s %d directories", m.getOPStr(), snap.dirNum)
		}
		strList = append(strList, str)
	}
	if snap.skipNum != 0 {
		strList = append(strList, fmt.Sprintf("skip %d %s", snap.skipNum, m.getSubject()))
	}
	return fmt.Sprintf("(%s)", strings.Join(strList, ", "))
}

func (m *CPMonitor) getSpeed(snap *CPMonitorSnap) float64 {
	return (float64(snap.incrementSize) / 1024) / (float64(snap.duration) * 1e-9)
}

func (m *CPMonitor) getOPStr() string {
	switch m.op {
	case operationTypePut:
		return "upload"
	case operationTypeGet:
		return "download"
	default:
		return "copy"
	}
}

func (m *CPMonitor) getDealSizeDetail(snap *CPMonitorSnap) string {
	return fmt.Sprintf(", OK size: %s", getSizeString(snap.dealSize))
}

func (m *CPMonitor) getSkipSize(snap *CPMonitorSnap) string {
	if snap.skipSize != 0 {
		return fmt.Sprintf(", Skip size: %s", getSizeString(snap.skipSize))
	}
	return ""
}

func (m *CPMonitor) getSizeDetail(snap *CPMonitorSnap) string {
	if snap.skipSize == 0 {
		return fmt.Sprintf(", Transfer size: %s", getSizeString(snap.transferSize))
	}
	if snap.transferSize == 0 {
		return fmt.Sprintf(", Skip size: %s", getSizeString(snap.skipSize))
	}
	return fmt.Sprintf(", OK size: %s(transfer: %s, skip: %s)", getSizeString(snap.transferSize+snap.skipSize), getSizeString(snap.transferSize), getSizeString(snap.skipSize))
}

func (m *CPMonitor) getPrecent(snap *CPMonitorSnap) float64 {
	if m.seekAheadEnd && m.seekAheadError == nil {
		if m.totalSize != 0 {
			return float64((snap.dealSize)*100.0) / float64(m.totalSize)
		}
		if m.totalNum != 0 {
			return float64((snap.dealNum)*100.0) / float64(m.totalNum)
		}
		return 100
	}
	return 0
}