LBRYCommunity/lbry.go
1
0
Fork 0
Code Issues 5 Pull requests Projects Releases 46 Packages Wiki Activity

fix slack logging

Browse source 
add failure cases
change interrupt behavior for fatal errors
bump up timeout to 20 minutes
change max file size to 2GB
remove unnecessary utxo wait
This commit is contained in:
Niko Storni 2018-06-15 17:03:28 -04:00
parent fb748cb255
commit d596240099
No known key found for this signature in database
GPG key ID: F37FE63398800368
7 changed files with 73 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
cmd/selfsync.go
View file

@ -144,7 +144,7 @@ func spaceCheck() error {
if usedPctile >= 0.90 && !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))
}
util.SendToSlackInfo("disk usage: %.1f%%", usedPctile*100)
util.SendInfoToSlack("disk usage: %.1f%%", usedPctile*100)
return nil
}
@ -153,11 +153,16 @@ func selfSync(cmd *cobra.Command, args []string) {
if slackToken == "" {
log.Error("A slack token was not present in env vars! Slack messages disabled!")
} else {
util.InitSlack(os.Getenv("SLACK_TOKEN"))
host, err := os.Hostname()
if err != nil {
log.Error("could not detect system hostname")
host = "ytsync-unknown"
}
util.InitSlack(os.Getenv("SLACK_TOKEN"), os.Getenv("SLACK_CHANNEL"), host)
}
err := spaceCheck()
if err != nil {
util.SendToSlackError(err.Error())
util.SendErrorToSlack(err.Error())
return
}
ytAPIKey := args[0]
@ -198,11 +203,11 @@ func selfSync(cmd *cobra.Command, args []string) {
for _, v := range queuesToSync {
interruptedByUser, err := processQueue(v, ytAPIKey, &syncCount)
if err != nil {
util.SendToSlackError(err.Error())
util.SendErrorToSlack(err.Error())
break mainLoop
}
if interruptedByUser {
util.SendToSlackInfo("interrupted by user!")
util.SendInfoToSlack("interrupted by user!")
break mainLoop
}
}
@ -211,15 +216,15 @@ func selfSync(cmd *cobra.Command, args []string) {
// sync whatever was specified
_, err := processQueue(syncStatus, ytAPIKey, &syncCount)
if err != nil {
util.SendToSlackError(err.Error())
util.SendErrorToSlack(err.Error())
return
}
}
util.SendToSlackInfo("Syncing process terminated!")
util.SendInfoToSlack("Syncing process terminated!")
}
func processQueue(queueStatus string, ytAPIKey string, syncCount *int) (bool, error) {
util.SendToSlackInfo("Syncing %s channels", queueStatus)
util.SendInfoToSlack("Syncing %s channels", queueStatus)
channelsToSync, err := fetchChannels(queueStatus)
if err != nil {
return false, errors.Prefix("failed to fetch channels", err)
@ -237,7 +242,7 @@ func processQueue(queueStatus string, ytAPIKey string, syncCount *int) (bool, er
}
}
interrupted, err := syncChannels(channelsToSync, ytAPIKey, syncCount)
util.SendToSlackInfo("Finished syncing %s channels", queueStatus)
util.SendInfoToSlack("Finished syncing %s channels", queueStatus)
return interrupted, err
}
@ -259,23 +264,23 @@ func syncChannels(channelsToSync []APIYoutubeChannel, ytAPIKey string, syncCount
channelID := channel.ChannelId
lbryChannelName := channel.DesiredChannelName
if channel.TotalVideos < 1 {
util.SendToSlackInfo("Channel %s has no videos. Skipping", lbryChannelName)
util.SendInfoToSlack("Channel %s has no videos. Skipping", lbryChannelName)
continue
}
if !channel.SyncServer.IsNull() && channel.SyncServer.String != host {
util.SendToSlackInfo("Channel %s is being synced by another server: %s", lbryChannelName, channel.SyncServer.String)
util.SendInfoToSlack("Channel %s is being synced by another server: %s", lbryChannelName, channel.SyncServer.String)
continue
}
//acquire the lock on the channel
err := setChannelSyncStatus(channelID, StatusSyncing)
if err != nil {
//util.SendToSlackError("Failed acquiring sync rights for channel %s: %v", lbryChannelName, err)
//util.SendErrorToSlack("Failed acquiring sync rights for channel %s: %v", lbryChannelName, err)
continue
}
//increment only if successfully acquired lock
*syncCount++
util.SendToSlackInfo("Syncing %s to LBRY! (iteration %d)", lbryChannelName, *syncCount)
util.SendInfoToSlack("Syncing %s to LBRY! (iteration %d)", lbryChannelName, *syncCount)
s := sync.Sync{
YoutubeAPIKey: ytAPIKey,
@ -289,15 +294,17 @@ func syncChannels(channelsToSync []APIYoutubeChannel, ytAPIKey string, syncCount
}
err = s.FullCycle()
util.SendToSlackInfo("Syncing " + lbryChannelName + " reached an end.")
util.SendInfoToSlack("Syncing " + lbryChannelName + " reached an end.")
if err != nil {
util.SendToSlackError(errors.FullTrace(err))
util.SendErrorToSlack(errors.FullTrace(err))
fatalErrors := []string{
"default_wallet already exists",
"WALLET HAS NOT BEEN MOVED TO THE WALLET BACKUP DIR",
"NotEnoughFunds",
"no space left on device",
}
if util.InSliceContains(err.Error(), fatalErrors) {
return s.IsInterrupted(), errors.Prefix("@Nikooo777 this requires manual intervention! Exiting...", err)
return true, errors.Prefix("@Nikooo777 this requires manual intervention! Exiting...", err)
}
//mark video as failed
err := setChannelSyncStatus(channelID, StatusFailed)

21
cmd/ytsync.go
View file

@ -31,23 +31,28 @@ func ytsync(cmd *cobra.Command, args []string) {
if slackToken == "" {
log.Error("A slack token was not present in env vars! Slack messages disabled!")
} else {
util.InitSlack(os.Getenv("SLACK_TOKEN"))
host, err := os.Hostname()
if err != nil {
log.Error("could not detect system hostname")
host = "ytsync-unknown"
}
util.InitSlack(os.Getenv("SLACK_TOKEN"), os.Getenv("SLACK_CHANNEL"), host)
}
usr, err := user.Current()
if err != nil {
util.SendToSlackError(err.Error())
util.SendErrorToSlack(err.Error())
return
}
usedPctile, err := util.GetUsedSpace(usr.HomeDir + "/.lbrynet/blobfiles/")
if err != nil {
util.SendToSlackError(err.Error())
util.SendErrorToSlack(err.Error())
return
}
if usedPctile > 0.9 && !skipSpaceCheck {
util.SendToSlackError("more than 90%% of the space has been used. use --skip-space-check to ignore. Used: %.1f%%", usedPctile*100)
util.SendErrorToSlack("more than 90%% of the space has been used. use --skip-space-check to ignore. Used: %.1f%%", usedPctile*100)
return
}
util.SendToSlackInfo("disk usage: %.1f%%", usedPctile*100)
util.SendInfoToSlack("disk usage: %.1f%%", usedPctile*100)
ytAPIKey := args[0]
lbryChannelName := args[1]
@ -69,7 +74,7 @@ func ytsync(cmd *cobra.Command, args []string) {
log.Errorln("setting --max-tries less than 1 doesn't make sense")
return
}
util.SendToSlackInfo("Syncing " + lbryChannelName + " to LBRY!")
util.SendInfoToSlack("Syncing " + lbryChannelName + " to LBRY!")
s := sync.Sync{
YoutubeAPIKey: ytAPIKey,
@ -85,7 +90,7 @@ func ytsync(cmd *cobra.Command, args []string) {
err = s.FullCycle()
if err != nil {
util.SendToSlackError(errors.FullTrace(err))
util.SendErrorToSlack(errors.FullTrace(err))
}
util.SendToSlackInfo("Syncing " + lbryChannelName + " reached an end.")
util.SendInfoToSlack("Syncing " + lbryChannelName + " reached an end.")
}

15
util/slack.go
View file

@ -1,6 +1,7 @@
package util
import (
"fmt"
"strings"
"github.com/lbryio/lbry.go/errors"
@ -46,6 +47,20 @@ func SendToSlack(message string) error {
return sendToSlack(defaultChannel, defaultUsername, message)
}
// SendErrorToSlack Sends an error message to the default channel and to the process log.
func SendErrorToSlack(format string, a ...interface{}) error {
message := fmt.Sprintf(format, a...)
log.Errorln(message)
return sendToSlack(defaultChannel, defaultUsername, ":sos: "+message)
}
// SendInfoToSlack Sends an info message to the default channel and to the process log.
func SendInfoToSlack(format string, a ...interface{}) error {
message := fmt.Sprintf(format, a...)
log.Debugln(message)
return sendToSlack(defaultChannel, defaultUsername, ":information_source: "+message)
}
func sendToSlack(channel, username, message string) error {
var err error

10
util/slack_test.go
View file

@ -10,7 +10,11 @@ func TestSendToSlack(t *testing.T) {
if slackToken == "" {
t.Error("A slack token was not provided")
}
InitSlack(slackToken)
SendToSlackInfo("This is a test :) Working %.2f%%", 1.01*100)
SendToSlackError("This is a test :) Working %.2f%%", 0.01*100)
host, err := os.Hostname()
if err != nil {
host = "ytsync-unknown"
}
InitSlack(os.Getenv("SLACK_TOKEN"), os.Getenv("SLACK_CHANNEL"), host)
SendInfoToSlack("This is a test :) Working %.2f%%", 1.01*100)
SendErrorToSlack("This is a test :) Working %.2f%%", 0.01*100)
}

12
ytsync/setup.go
View file

@ -41,6 +41,9 @@ func (s *Sync) walletSetup() error {
}
}
log.Debugf("Source channel has %d videos", numOnSource)
if numOnSource == 0 {
return nil
}
numPublished, err := s.daemon.NumClaimsInChannel(s.LbryChannelName)
if err != nil {
@ -50,7 +53,7 @@ func (s *Sync) walletSetup() error {
minBalance := (float64(numOnSource)-float64(numPublished))*(publishAmount+0.1) + channelClaimAmount
if numPublished > numOnSource {
util.SendToSlackError("something is going on as we published more videos than those available on source: %d/%d", numPublished, numOnSource)
util.SendErrorToSlack("something is going on as we published more videos than those available on source: %d/%d", numPublished, numOnSource)
minBalance = 1 //since we ended up in this function it means some juice is still needed
}
amountToAdd, _ := decimal.NewFromFloat(minBalance).Sub(balance).Float64()
@ -166,7 +169,7 @@ func (s *Sync) waitUntilUTXOsConfirmed() error {
if time.Now().After(origin.Add(15 * time.Minute)) {
//lbryum is messing with us or something. restart the daemon
//this could also be a very long block
util.SendToSlackError("We've been waiting UTXOs confirmation for %s... and this isn't normal", time.Now().Sub(origin).String())
util.SendErrorToSlack("We've been waiting UTXOs confirmation for %s... and this isn't normal", time.Now().Sub(origin).String())
}
wait := 30 * time.Second
log.Println("Waiting " + wait.String() + "...")
@ -282,6 +285,7 @@ func (s *Sync) addCredits(amountToAdd float64) error {
log.Println("Waiting " + wait.String() + " for lbryum to let us know we have the new transaction")
time.Sleep(wait)
log.Println("Waiting for transaction to be confirmed")
return s.waitUntilUTXOsConfirmed()
return nil
//log.Println("Waiting for transaction to be confirmed")
//return s.waitUntilUTXOsConfirmed()
}

2
ytsync/sources/youtubeVideo.go
View file

@ -211,7 +211,7 @@ func (v YoutubeVideo) Sync(daemon *jsonrpc.Client, claimAddress string, amount f
if err != nil {
return err
}
if fi.Size() > 1*1024*1024*1024 {
if fi.Size() > 2*1024*1024*1024 {
//delete the video and ignore the error
_ = v.delete()
return errors.Err("video is bigger than 1GB, skipping for now")

11
ytsync/ytsync.go
View file

@ -159,7 +159,7 @@ func (s *Sync) FullCycle() error {
log.Infoln("Waiting for daemon to finish starting...")
s.daemon = jsonrpc.NewClient("")
s.daemon.SetRPCTimeout(5 * time.Minute)
s.daemon.SetRPCTimeout(20 * time.Minute)
WaitForDaemonStart:
for {
@ -188,8 +188,8 @@ WaitForDaemonStart:
return nil
}
func logShutdownError(shutdownErr error) {
util.SendToSlackError("error shutting down daemon: %v", shutdownErr)
util.SendToSlackError("WALLET HAS NOT BEEN MOVED TO THE WALLET BACKUP DIR")
util.SendErrorToSlack("error shutting down daemon: %v", shutdownErr)
util.SendErrorToSlack("WALLET HAS NOT BEEN MOVED TO THE WALLET BACKUP DIR")
}
func (s *Sync) doSync() error {
@ -256,6 +256,7 @@ func (s *Sync) startWorker(workerNum int) {
fatalErrors := []string{
":5279: read: connection reset by peer",
"no space left on device",
"NotEnoughFunds",
}
if util.InSliceContains(err.Error(), fatalErrors) || s.StopOnError {
s.grp.Stop()
@ -283,14 +284,14 @@ func (s *Sync) startWorker(workerNum int) {
err = s.walletSetup()
if err != nil {
s.stop.Stop()
util.SendToSlackError("Failed to setup the wallet for a refill: %v", err)
util.SendErrorToSlack("Failed to setup the wallet for a refill: %v", err)
break
}
}
log.Println("Retrying")
continue
}
util.SendToSlackError("Video failed after %d retries, skipping. Stack: %s", tryCount, logMsg)
util.SendErrorToSlack("Video failed after %d retries, skipping. Stack: %s", tryCount, logMsg)
}
}
break