ytsync/manager/manager.go

193 lines
6 KiB
Go
Raw Normal View History

package manager
import (
"fmt"
"strings"
"syscall"
"time"
2020-06-11 18:45:56 +02:00
"github.com/lbryio/ytsync/v5/blobs_reflector"
"github.com/lbryio/ytsync/v5/ip_manager"
"github.com/lbryio/ytsync/v5/namer"
"github.com/lbryio/ytsync/v5/sdk"
2020-08-08 01:12:55 +02:00
"github.com/lbryio/ytsync/v5/shared"
2020-06-11 18:45:56 +02:00
logUtils "github.com/lbryio/ytsync/v5/util"
"github.com/lbryio/lbry.go/v2/extras/errors"
"github.com/lbryio/lbry.go/v2/extras/util"
2019-06-06 23:25:31 +02:00
log "github.com/sirupsen/logrus"
)
type SyncManager struct {
2020-08-08 01:12:55 +02:00
CliFlags shared.SyncFlags
ApiConfig *sdk.APIConfig
LbrycrdDsn string
AwsConfigs *shared.AwsConfigs
blobsDir string
channelsToSync []Sync
2018-09-18 21:20:34 +02:00
}
2020-08-08 01:12:55 +02:00
func NewSyncManager(cliFlags shared.SyncFlags, blobsDir, lbrycrdDsn string, awsConfigs *shared.AwsConfigs, apiConfig *sdk.APIConfig) *SyncManager {
2018-09-18 21:20:34 +02:00
return &SyncManager{
2020-08-08 01:12:55 +02:00
CliFlags: cliFlags,
blobsDir: blobsDir,
LbrycrdDsn: lbrycrdDsn,
AwsConfigs: awsConfigs,
ApiConfig: apiConfig,
2018-09-18 21:20:34 +02:00
}
}
2020-08-08 01:12:55 +02:00
func (s *SyncManager) enqueueChannel(channel *shared.YoutubeChannel) {
s.channelsToSync = append(s.channelsToSync, Sync{
DbChannelData: channel,
Manager: s,
namer: namer.NewNamer(),
})
}
func (s *SyncManager) Start() error {
2019-08-13 05:20:09 +02:00
if logUtils.ShouldCleanOnStartup() {
err := logUtils.CleanForStartup()
if err != nil {
return err
}
}
var lastChannelProcessed string
syncCount := 0
for {
2020-08-08 01:12:55 +02:00
s.channelsToSync = make([]Sync, 0, 10) // reset sync queue
err := s.checkUsedSpace()
if err != nil {
return errors.Err(err)
}
shouldInterruptLoop := false
2020-08-08 01:12:55 +02:00
if s.CliFlags.IsSingleChannelSync() {
channels, err := s.ApiConfig.FetchChannels("", &s.CliFlags)
if err != nil {
return errors.Err(err)
}
if len(channels) != 1 {
return errors.Err("Expected 1 channel, %d returned", len(channels))
}
2020-08-08 01:12:55 +02:00
s.enqueueChannel(&channels[0])
shouldInterruptLoop = true
} else {
var queuesToSync []string
2020-08-08 01:12:55 +02:00
if s.CliFlags.SyncStatus != "" {
queuesToSync = append(queuesToSync, s.CliFlags.SyncStatus)
} else if s.CliFlags.SyncUpdate {
queuesToSync = append(queuesToSync, shared.StatusSyncing, shared.StatusSynced)
} else {
2020-08-08 01:12:55 +02:00
queuesToSync = append(queuesToSync, shared.StatusSyncing, shared.StatusQueued)
}
2019-08-20 11:33:06 +02:00
queues:
for _, q := range queuesToSync {
2020-08-08 01:12:55 +02:00
channels, err := s.ApiConfig.FetchChannels(q, &s.CliFlags)
if err != nil {
return err
}
2020-08-08 01:12:55 +02:00
log.Infof("Currently processing the \"%s\" queue with %d channels", q, len(channels))
for _, c := range channels {
s.enqueueChannel(&c)
queueAll := q == shared.StatusFailed || q == shared.StatusSyncing
if !queueAll {
break queues
2019-08-20 11:33:06 +02:00
}
}
2020-08-08 01:12:55 +02:00
log.Infof("Drained the \"%s\" queue", q)
}
}
2020-08-08 01:12:55 +02:00
if len(s.channelsToSync) == 0 {
log.Infoln("No channels to sync. Pausing 5 minutes!")
time.Sleep(5 * time.Minute)
}
2020-08-08 01:12:55 +02:00
for _, sync := range s.channelsToSync {
if lastChannelProcessed == sync.DbChannelData.ChannelId {
util.SendToSlack("We just killed a sync for %s to stop looping! (%s)", sync.DbChannelData.DesiredChannelName, sync.DbChannelData.ChannelId)
stopTheLoops := errors.Err("Found channel %s running twice, set it to failed, and reprocess later", sync.DbChannelData.DesiredChannelName)
sync.setChannelTerminationStatus(&stopTheLoops)
continue
}
2020-08-08 01:12:55 +02:00
lastChannelProcessed = sync.DbChannelData.ChannelId
shouldNotCount := false
2020-08-08 01:12:55 +02:00
logUtils.SendInfoToSlack("Syncing %s (%s) to LBRY! total processed channels since startup: %d", sync.DbChannelData.DesiredChannelName, sync.DbChannelData.ChannelId, syncCount+1)
err := sync.FullCycle()
//TODO: THIS IS A TEMPORARY WORK AROUND FOR THE STUPID IP LOCKUP BUG
ipPool, _ := ip_manager.GetIPPool(sync.grp)
if ipPool != nil {
ipPool.ReleaseAll()
}
if err != nil {
if strings.Contains(err.Error(), "quotaExceeded") {
logUtils.SleepUntilQuotaReset()
}
fatalErrors := []string{
"default_wallet already exists",
"WALLET HAS NOT BEEN MOVED TO THE WALLET BACKUP DIR",
"NotEnoughFunds",
"no space left on device",
"failure uploading wallet",
"the channel in the wallet is different than the channel in the database",
"this channel does not belong to this wallet!",
"You already have a stream claim published under the name",
}
2020-05-30 02:18:20 +02:00
2018-07-24 02:01:35 +02:00
if util.SubstringInSlice(err.Error(), fatalErrors) {
return errors.Prefix("@Nikooo777 this requires manual intervention! Exiting...", err)
}
shouldNotCount = strings.Contains(err.Error(), "this youtube channel is being managed by another server")
2018-08-22 16:56:02 +02:00
if !shouldNotCount {
logUtils.SendInfoToSlack("A non fatal error was reported by the sync process.\n%s", errors.FullTrace(err))
2018-08-22 16:56:02 +02:00
}
}
err = blobs_reflector.ReflectAndClean()
if err != nil {
return errors.Prefix("@Nikooo777 something went wrong while reflecting blobs", err)
}
2020-08-08 01:12:55 +02:00
logUtils.SendInfoToSlack("%s (%s) reached an end. Total processed channels since startup: %d", sync.DbChannelData.DesiredChannelName, sync.DbChannelData.ChannelId, syncCount+1)
if !shouldNotCount {
syncCount++
}
2020-08-08 01:12:55 +02:00
if sync.IsInterrupted() || (s.CliFlags.Limit != 0 && syncCount >= s.CliFlags.Limit) {
shouldInterruptLoop = true
break
}
}
2020-08-08 01:12:55 +02:00
if shouldInterruptLoop || s.CliFlags.SingleRun {
break
}
}
return nil
}
2020-08-08 01:12:55 +02:00
func (s *SyncManager) checkUsedSpace() error {
usedPctile, err := GetUsedSpace(logUtils.GetBlobsDir())
if err != nil {
return errors.Err(err)
}
2020-08-08 01:12:55 +02:00
if usedPctile >= 0.90 && !s.CliFlags.SkipSpaceCheck {
return errors.Err(fmt.Sprintf("more than 90%% of the space has been used. use --skip-space-check to ignore. Used: %.1f%%", usedPctile*100))
}
2018-08-21 20:07:30 +02:00
log.Infof("disk usage: %.1f%%", usedPctile*100)
return nil
}
// GetUsedSpace returns a value between 0 and 1, with 0 being completely empty and 1 being full, for the disk that holds the provided path
func GetUsedSpace(path string) (float32, error) {
var stat syscall.Statfs_t
err := syscall.Statfs(path, &stat)
if err != nil {
return 0, err
}
// Available blocks * size per block = available space in bytes
all := stat.Blocks * uint64(stat.Bsize)
free := stat.Bfree * uint64(stat.Bsize)
used := all - free
return float32(used) / float32(all), nil
}