mirror of
https://github.com/rclone/rclone.git
synced 2024-12-13 02:33:58 +01:00
435 lines
10 KiB
Go
435 lines
10 KiB
Go
package accounting
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/rclone/rclone/fs/rc"
|
|
|
|
"github.com/rclone/rclone/fs"
|
|
)
|
|
|
|
const globalStats = "global_stats"
|
|
|
|
var groups *statsGroups
|
|
|
|
func init() {
|
|
// Init stats container
|
|
groups = newStatsGroups()
|
|
|
|
// Set the function pointer up in fs
|
|
fs.CountError = GlobalStats().Error
|
|
}
|
|
|
|
func rcListStats(ctx context.Context, in rc.Params) (rc.Params, error) {
|
|
out := make(rc.Params)
|
|
|
|
out["groups"] = groups.names()
|
|
|
|
return out, nil
|
|
}
|
|
|
|
func init() {
|
|
rc.Add(rc.Call{
|
|
Path: "core/group-list",
|
|
Fn: rcListStats,
|
|
Title: "Returns list of stats.",
|
|
Help: `
|
|
This returns list of stats groups currently in memory.
|
|
|
|
Returns the following values:
|
|
` + "```" + `
|
|
{
|
|
"groups": an array of group names:
|
|
[
|
|
"group1",
|
|
"group2",
|
|
...
|
|
]
|
|
}
|
|
` + "```" + `
|
|
`,
|
|
})
|
|
}
|
|
|
|
func rcRemoteStats(ctx context.Context, in rc.Params) (rc.Params, error) {
|
|
// Check to see if we should filter by group.
|
|
group, err := in.GetString("group")
|
|
if rc.NotErrParamNotFound(err) {
|
|
return rc.Params{}, err
|
|
}
|
|
if group != "" {
|
|
return StatsGroup(ctx, group).RemoteStats()
|
|
}
|
|
|
|
return groups.sum(ctx).RemoteStats()
|
|
}
|
|
|
|
func init() {
|
|
rc.Add(rc.Call{
|
|
Path: "core/stats",
|
|
Fn: rcRemoteStats,
|
|
Title: "Returns stats about current transfers.",
|
|
Help: `
|
|
This returns all available stats:
|
|
|
|
rclone rc core/stats
|
|
|
|
If group is not provided then summed up stats for all groups will be
|
|
returned.
|
|
|
|
Parameters
|
|
|
|
- group - name of the stats group (string)
|
|
|
|
Returns the following values:
|
|
|
|
` + "```" + `
|
|
{
|
|
"bytes": total transferred bytes since the start of the group,
|
|
"checks": number of files checked,
|
|
"deletes" : number of files deleted,
|
|
"elapsedTime": time in floating point seconds since rclone was started,
|
|
"errors": number of errors,
|
|
"eta": estimated time in seconds until the group completes,
|
|
"fatalError": boolean whether there has been at least one fatal error,
|
|
"lastError": last error string,
|
|
"renames" : number of files renamed,
|
|
"retryError": boolean showing whether there has been at least one non-NoRetryError,
|
|
"speed": average speed in bytes per second since start of the group,
|
|
"totalBytes": total number of bytes in the group,
|
|
"totalChecks": total number of checks in the group,
|
|
"totalTransfers": total number of transfers in the group,
|
|
"transferTime" : total time spent on running jobs,
|
|
"transfers": number of transferred files,
|
|
"transferring": an array of currently active file transfers:
|
|
[
|
|
{
|
|
"bytes": total transferred bytes for this file,
|
|
"eta": estimated time in seconds until file transfer completion
|
|
"name": name of the file,
|
|
"percentage": progress of the file transfer in percent,
|
|
"speed": average speed over the whole transfer in bytes per second,
|
|
"speedAvg": current speed in bytes per second as an exponentially weighted moving average,
|
|
"size": size of the file in bytes
|
|
}
|
|
],
|
|
"checking": an array of names of currently active file checks
|
|
[]
|
|
}
|
|
` + "```" + `
|
|
Values for "transferring", "checking" and "lastError" are only assigned if data is available.
|
|
The value for "eta" is null if an eta cannot be determined.
|
|
`,
|
|
})
|
|
}
|
|
|
|
func rcTransferredStats(ctx context.Context, in rc.Params) (rc.Params, error) {
|
|
// Check to see if we should filter by group.
|
|
group, err := in.GetString("group")
|
|
if rc.NotErrParamNotFound(err) {
|
|
return rc.Params{}, err
|
|
}
|
|
|
|
out := make(rc.Params)
|
|
if group != "" {
|
|
out["transferred"] = StatsGroup(ctx, group).Transferred()
|
|
} else {
|
|
out["transferred"] = groups.sum(ctx).Transferred()
|
|
}
|
|
|
|
return out, nil
|
|
}
|
|
|
|
func init() {
|
|
rc.Add(rc.Call{
|
|
Path: "core/transferred",
|
|
Fn: rcTransferredStats,
|
|
Title: "Returns stats about completed transfers.",
|
|
Help: `
|
|
This returns stats about completed transfers:
|
|
|
|
rclone rc core/transferred
|
|
|
|
If group is not provided then completed transfers for all groups will be
|
|
returned.
|
|
|
|
Note only the last 100 completed transfers are returned.
|
|
|
|
Parameters
|
|
|
|
- group - name of the stats group (string)
|
|
|
|
Returns the following values:
|
|
` + "```" + `
|
|
{
|
|
"transferred": an array of completed transfers (including failed ones):
|
|
[
|
|
{
|
|
"name": name of the file,
|
|
"size": size of the file in bytes,
|
|
"bytes": total transferred bytes for this file,
|
|
"checked": if the transfer is only checked (skipped, deleted),
|
|
"timestamp": integer representing millisecond unix epoch,
|
|
"error": string description of the error (empty if successful),
|
|
"jobid": id of the job that this transfer belongs to
|
|
}
|
|
]
|
|
}
|
|
` + "```" + `
|
|
`,
|
|
})
|
|
}
|
|
|
|
func rcResetStats(ctx context.Context, in rc.Params) (rc.Params, error) {
|
|
// Check to see if we should filter by group.
|
|
group, err := in.GetString("group")
|
|
if rc.NotErrParamNotFound(err) {
|
|
return rc.Params{}, err
|
|
}
|
|
|
|
if group != "" {
|
|
stats := groups.get(group)
|
|
if stats == nil {
|
|
return rc.Params{}, fmt.Errorf("group %q not found", group)
|
|
}
|
|
stats.ResetErrors()
|
|
stats.ResetCounters()
|
|
} else {
|
|
groups.reset()
|
|
}
|
|
|
|
return rc.Params{}, nil
|
|
}
|
|
|
|
func init() {
|
|
rc.Add(rc.Call{
|
|
Path: "core/stats-reset",
|
|
Fn: rcResetStats,
|
|
Title: "Reset stats.",
|
|
Help: `
|
|
This clears counters, errors and finished transfers for all stats or specific
|
|
stats group if group is provided.
|
|
|
|
Parameters
|
|
|
|
- group - name of the stats group (string)
|
|
`,
|
|
})
|
|
}
|
|
|
|
func rcDeleteStats(ctx context.Context, in rc.Params) (rc.Params, error) {
|
|
// Group name required because we only do single group.
|
|
group, err := in.GetString("group")
|
|
if rc.NotErrParamNotFound(err) {
|
|
return rc.Params{}, err
|
|
}
|
|
|
|
if group != "" {
|
|
groups.delete(group)
|
|
}
|
|
|
|
return rc.Params{}, nil
|
|
}
|
|
|
|
func init() {
|
|
rc.Add(rc.Call{
|
|
Path: "core/stats-delete",
|
|
Fn: rcDeleteStats,
|
|
Title: "Delete stats group.",
|
|
Help: `
|
|
This deletes entire stats group.
|
|
|
|
Parameters
|
|
|
|
- group - name of the stats group (string)
|
|
`,
|
|
})
|
|
}
|
|
|
|
type statsGroupCtx int64
|
|
|
|
const statsGroupKey statsGroupCtx = 1
|
|
|
|
// WithStatsGroup returns copy of the parent context with assigned group.
|
|
func WithStatsGroup(parent context.Context, group string) context.Context {
|
|
return context.WithValue(parent, statsGroupKey, group)
|
|
}
|
|
|
|
// StatsGroupFromContext returns group from the context if it's available.
|
|
// Returns false if group is empty.
|
|
func StatsGroupFromContext(ctx context.Context) (string, bool) {
|
|
statsGroup, ok := ctx.Value(statsGroupKey).(string)
|
|
if statsGroup == "" {
|
|
ok = false
|
|
}
|
|
return statsGroup, ok
|
|
}
|
|
|
|
// Stats gets stats by extracting group from context.
|
|
func Stats(ctx context.Context) *StatsInfo {
|
|
group, ok := StatsGroupFromContext(ctx)
|
|
if !ok {
|
|
return GlobalStats()
|
|
}
|
|
return StatsGroup(ctx, group)
|
|
}
|
|
|
|
// StatsGroup gets stats by group name.
|
|
func StatsGroup(ctx context.Context, group string) *StatsInfo {
|
|
stats := groups.get(group)
|
|
if stats == nil {
|
|
return NewStatsGroup(ctx, group)
|
|
}
|
|
return stats
|
|
}
|
|
|
|
// GlobalStats returns special stats used for global accounting.
|
|
func GlobalStats() *StatsInfo {
|
|
return StatsGroup(context.Background(), globalStats)
|
|
}
|
|
|
|
// NewStatsGroup creates new stats under named group.
|
|
func NewStatsGroup(ctx context.Context, group string) *StatsInfo {
|
|
stats := NewStats(ctx)
|
|
stats.group = group
|
|
groups.set(ctx, group, stats)
|
|
return stats
|
|
}
|
|
|
|
// statsGroups holds a synchronized map of stats
|
|
type statsGroups struct {
|
|
mu sync.Mutex
|
|
m map[string]*StatsInfo
|
|
order []string
|
|
}
|
|
|
|
// newStatsGroups makes a new statsGroups object
|
|
func newStatsGroups() *statsGroups {
|
|
return &statsGroups{
|
|
m: make(map[string]*StatsInfo),
|
|
}
|
|
}
|
|
|
|
// set marks the stats as belonging to a group
|
|
func (sg *statsGroups) set(ctx context.Context, group string, stats *StatsInfo) {
|
|
sg.mu.Lock()
|
|
defer sg.mu.Unlock()
|
|
ci := fs.GetConfig(ctx)
|
|
|
|
// Limit number of groups kept in memory.
|
|
if len(sg.order) >= ci.MaxStatsGroups {
|
|
group := sg.order[0]
|
|
fs.LogPrintf(fs.LogLevelDebug, nil, "Max number of stats groups reached removing %s", group)
|
|
delete(sg.m, group)
|
|
r := (len(sg.order) - ci.MaxStatsGroups) + 1
|
|
sg.order = sg.order[r:]
|
|
}
|
|
|
|
// Exclude global stats from listing
|
|
if group != globalStats {
|
|
sg.order = append(sg.order, group)
|
|
}
|
|
sg.m[group] = stats
|
|
}
|
|
|
|
// get gets the stats for group, or nil if not found
|
|
func (sg *statsGroups) get(group string) *StatsInfo {
|
|
sg.mu.Lock()
|
|
defer sg.mu.Unlock()
|
|
stats, ok := sg.m[group]
|
|
if !ok {
|
|
return nil
|
|
}
|
|
return stats
|
|
}
|
|
|
|
func (sg *statsGroups) names() []string {
|
|
sg.mu.Lock()
|
|
defer sg.mu.Unlock()
|
|
return sg.order
|
|
}
|
|
|
|
// sum returns aggregate stats that contains summation of all groups.
|
|
func (sg *statsGroups) sum(ctx context.Context) *StatsInfo {
|
|
startTime := GlobalStats().startTime
|
|
sg.mu.Lock()
|
|
defer sg.mu.Unlock()
|
|
|
|
sum := NewStats(ctx)
|
|
for _, stats := range sg.m {
|
|
stats.mu.RLock()
|
|
{
|
|
sum.bytes += stats.bytes
|
|
sum.errors += stats.errors
|
|
if sum.lastError == nil && stats.lastError != nil {
|
|
sum.lastError = stats.lastError
|
|
}
|
|
sum.fatalError = sum.fatalError || stats.fatalError
|
|
sum.retryError = sum.retryError || stats.retryError
|
|
if stats.retryAfter.After(sum.retryAfter) {
|
|
// Update the retryAfter field only if it is a later date than the current one in the sum
|
|
sum.retryAfter = stats.retryAfter
|
|
}
|
|
sum.checks += stats.checks
|
|
sum.checking.merge(stats.checking)
|
|
sum.checkQueue += stats.checkQueue
|
|
sum.checkQueueSize += stats.checkQueueSize
|
|
sum.transfers += stats.transfers
|
|
sum.transferring.merge(stats.transferring)
|
|
sum.transferQueueSize += stats.transferQueueSize
|
|
sum.renames += stats.renames
|
|
sum.renameQueue += stats.renameQueue
|
|
sum.renameQueueSize += stats.renameQueueSize
|
|
sum.deletes += stats.deletes
|
|
sum.deletedDirs += stats.deletedDirs
|
|
sum.inProgress.merge(stats.inProgress)
|
|
sum.startedTransfers = append(sum.startedTransfers, stats.startedTransfers...)
|
|
sum.oldTimeRanges = append(sum.oldTimeRanges, stats.oldTimeRanges...)
|
|
sum.oldDuration += stats.oldDuration
|
|
stats.average.mu.Lock()
|
|
sum.average.speed += stats.average.speed
|
|
stats.average.mu.Unlock()
|
|
}
|
|
stats.mu.RUnlock()
|
|
}
|
|
sum.startTime = startTime
|
|
return sum
|
|
}
|
|
|
|
func (sg *statsGroups) reset() {
|
|
sg.mu.Lock()
|
|
defer sg.mu.Unlock()
|
|
|
|
for _, stats := range sg.m {
|
|
stats.ResetErrors()
|
|
stats.ResetCounters()
|
|
}
|
|
|
|
sg.m = make(map[string]*StatsInfo)
|
|
sg.order = nil
|
|
}
|
|
|
|
// delete removes all references to the group.
|
|
func (sg *statsGroups) delete(group string) {
|
|
sg.mu.Lock()
|
|
defer sg.mu.Unlock()
|
|
stats := sg.m[group]
|
|
if stats == nil {
|
|
return
|
|
}
|
|
stats.ResetErrors()
|
|
stats.ResetCounters()
|
|
delete(sg.m, group)
|
|
|
|
// Remove group reference from the ordering slice.
|
|
tmp := sg.order[:0]
|
|
for _, g := range sg.order {
|
|
if g != group {
|
|
tmp = append(tmp, g)
|
|
}
|
|
}
|
|
sg.order = tmp
|
|
}
|