More general work towards zero downtime restarts, fairly stable now

autorole/bot.go

@@ -28,7 +28,7 @@ func (p *Plugin) BotInit() {
 	eventsystem.AddHandler(bot.RedisWrapper(OnMemberJoin), eventsystem.EventGuildMemberAdd)
 	eventsystem.AddHandler(HandlePresenceUpdate, eventsystem.EventPresenceUpdate)
 
-	pubsub.AddHandler("autorole_stop_processing", HandleUpdateAutomodRules, nil)
+	pubsub.AddHandler("autorole_stop_processing", HandleUpdateAutoroles, nil)
 }
 
 func (p *Plugin) BotStarted() {
@@ -49,7 +49,7 @@ var roleCommands = []*commands.YAGCommand{
 }
 
 // Stop updating
-func HandleUpdateAutomodRules(event *pubsub.Event) {
+func HandleUpdateAutoroles(event *pubsub.Event) {
 	stopProcessing(event.TargetGuildInt)
 }
 

bot/bot.go

@@ -127,7 +127,7 @@ func Run() {
 		state = StateWaitingHelloMaster
 		stateLock.Unlock()
 
-		log.Println("Attempting to connect to master at ", masterAddr)
+		log.Println("Connecting to master at ", masterAddr, ", wont start until connected and told to start")
 		SlaveClient, err = slave.ConnectToMaster(&SlaveImpl{}, masterAddr)
 		if err != nil {
 			log.WithError(err).Error("Failed connecting to master")

bot/plugin.go

@@ -2,6 +2,7 @@ package bot
 
 import (
 	"github.com/jonas747/discordgo"
+	"github.com/jonas747/dutil/dstate"
 	"github.com/jonas747/yagpdb/common"
 	"github.com/mediocregopher/radix.v2/redis"
 	"github.com/sirupsen/logrus"
@@ -37,6 +38,10 @@ type BotStopperHandler interface {
 	StopBot(wg *sync.WaitGroup)
 }
 
+type ShardMigrationHandler interface {
+	GuildMigrated(guild *dstate.GuildState, toThisSlave bool)
+}
+
 func EmitGuildRemoved(guildID int64) {
 	client := common.MustGetRedisClient()
 	defer common.RedisPool.Put(client)

bot/slaveimpl.go

@@ -2,6 +2,7 @@ package bot
 
 import (
 	"github.com/jonas747/dutil/dstate"
+	"github.com/jonas747/yagpdb/common"
 	"github.com/jonas747/yagpdb/master"
 	"github.com/jonas747/yagpdb/master/slave"
 	"github.com/sirupsen/logrus"
@@ -154,4 +155,12 @@ func (s *SlaveImpl) LoadGuildState(gs *master.GuildStateData) {
 		State.Channels[c.ID] = c
 	}
 	State.Unlock()
+
+	for _, v := range common.Plugins {
+		if guildMigrationHandler, ok := v.(ShardMigrationHandler); ok {
+			if ok {
+				guildMigrationHandler.GuildMigrated(guild, true)
+			}
+		}
+	}
 }

cmd/yagmaster/README.md

@@ -0,0 +1 @@
+Yagmaster manages the zero downtime restarts of the bot process, it can only be used if the processes are just running with `-bot`, since at some point there will be 2 processes running and you will run into issues with 2 processes trying to listen on the same port¨.

commands/commands.go

@@ -6,7 +6,6 @@ package commands
 import (
 	"github.com/jonas747/dcmd"
 	"github.com/jonas747/discordgo"
-	"github.com/jonas747/yagpdb/bot"
 	"github.com/jonas747/yagpdb/common"
 	"github.com/mediocregopher/radix.v2/redis"
 	log "github.com/sirupsen/logrus"
@@ -53,7 +52,6 @@ func InitCommands() {
 
 		ResponseSender: &dcmd.StdResponseSender{LogErrors: true},
 		Prefix:         &Plugin{},
-		State:          bot.State,
 	}
 
 	// We have our own middleware before the argument parsing, this is to check for things such as wether the command is enabled at all

commands/plugin_bot.go

@@ -22,6 +22,8 @@ var _ bot.BotInitHandler = (*Plugin)(nil)
 func (p *Plugin) BotInit() {
 	eventsystem.AddHandler(bot.RedisWrapper(HandleGuildCreate), eventsystem.EventGuildCreate)
 	eventsystem.AddHandler(handleMsgCreate, eventsystem.EventMessageCreate)
+
+	CommandSystem.State = bot.State
 }
 
 func YAGCommandMiddleware(inner dcmd.RunFunc) dcmd.RunFunc {

customcommands/bot.go

@@ -19,6 +19,17 @@ import (
 	"unicode/utf8"
 )
 
+var _ bot.BotInitHandler = (*Plugin)(nil)
+var _ commands.CommandProvider = (*Plugin)(nil)
+
+func (p *Plugin) AddCommands() {
+	commands.AddRootCommands(cmdListCommands)
+}
+
+func (p *Plugin) BotInit() {
+	eventsystem.AddHandler(bot.RedisWrapper(HandleMessageCreate), eventsystem.EventMessageCreate)
+}
+
 var cmdListCommands = &commands.YAGCommand{
 	CmdCategory:    commands.CategoryTool,
 	Name:           "CustomCommands",

customcommands/customcommands.go

@@ -4,9 +4,6 @@ import (
 	"encoding/json"
 	"github.com/jonas747/discordgo"
 	"github.com/jonas747/dutil/dstate"
-	"github.com/jonas747/yagpdb/bot"
-	"github.com/jonas747/yagpdb/bot/eventsystem"
-	"github.com/jonas747/yagpdb/commands"
 	"github.com/jonas747/yagpdb/common"
 	"github.com/karlseguin/ccache"
 	"github.com/mediocregopher/radix.v2/redis"
@@ -25,9 +22,6 @@ var (
 
 func KeyCommands(guildID int64) string { return "custom_commands:" + discordgo.StrID(guildID) }
 
-var _ bot.BotInitHandler = (*Plugin)(nil)
-var _ commands.CommandProvider = (*Plugin)(nil)
-
 type Plugin struct{}
 
 func RegisterPlugin() {
@@ -36,14 +30,6 @@ func RegisterPlugin() {
 	RegexCache = ccache.New(ccache.Configure())
 }
 
-func (p *Plugin) AddCommands() {
-	commands.AddRootCommands(cmdListCommands)
-}
-
-func (p *Plugin) BotInit() {
-	eventsystem.AddHandler(bot.RedisWrapper(HandleMessageCreate), eventsystem.EventMessageCreate)
-}
-
 func (p *Plugin) Name() string {
 	return "Custom commands"
 }

logs/plugin_bot.go

@@ -51,9 +51,7 @@ func (p *Plugin) BotInit() {
 	if err != nil {
 		panic("Failed preparing statement: " + err.Error())
 	}
-}
 
-func (p *Plugin) StartBot() {
 	go EvtProcesser()
 	go EvtProcesserGCs()
 }
@@ -280,7 +278,7 @@ var cmdNicknames = &commands.YAGCommand{
 	RunFunc: func(parsed *dcmd.Data) (interface{}, error) {
 		config, err := GetConfig(parsed.GS.ID)
 		if err != nil {
-			return "AAAAA", err
+			return "Failed retrieving config", err
 		}
 
 		target := parsed.Msg.Author

master/slave/slave.go

@@ -15,14 +15,7 @@ func ConnectToMaster(bot Bot, addr string) (*Conn, error) {
 		address: addr,
 	}
 
-	err := conn.Connect()
-	if err != nil {
-		return nil, err
-	}
-
-	conn.baseConn.SendNoLock(master.EvtSlaveHello, master.SlaveHelloData{
-		Running: false,
-	})
+	go conn.ReconnectLoop(false)
 
 	return conn, nil
 }
@@ -61,6 +54,10 @@ func (c *Conn) Connect() error {
 }
 
 func (c *Conn) OnClosedConn() {
+	go c.ReconnectLoop(true)
+}
+
+func (c *Conn) ReconnectLoop(running bool) {
 	c.mu.Lock()
 	if c.reconnecting {
 		c.mu.Unlock()
@@ -72,23 +69,23 @@ func (c *Conn) OnClosedConn() {
 
 	go func() {
 		for {
-			if c.TryReconnect() {
+			if c.TryReconnect(running) {
 				break
 			}
 
-			time.Sleep(time.Second)
+			time.Sleep(time.Second * 5)
 		}
 	}()
 }
 
-func (c *Conn) TryReconnect() bool {
+func (c *Conn) TryReconnect(running bool) bool {
 	err := c.Connect()
 	if err != nil {
 		return false
 	}
 
 	c.mu.Lock()
-	c.baseConn.SendNoLock(master.EvtSlaveHello, master.SlaveHelloData{Running: true})
+	c.baseConn.SendNoLock(master.EvtSlaveHello, master.SlaveHelloData{Running: running})
 	for _, v := range c.sendQueue {
 		c.baseConn.SendNoLock(v.EvtID, v.Data)
 	}

moderation/plugin_bot.go

@@ -7,6 +7,7 @@ import (
 	"github.com/jinzhu/gorm"
 	"github.com/jonas747/dcmd"
 	"github.com/jonas747/discordgo"
+	"github.com/jonas747/dutil/dstate"
 	"github.com/jonas747/yagpdb/bot"
 	"github.com/jonas747/yagpdb/bot/eventsystem"
 	"github.com/jonas747/yagpdb/commands"
@@ -34,6 +35,7 @@ const MuteDeniedChannelPerms = discordgo.PermissionSendMessages | discordgo.Perm
 
 var _ commands.CommandProvider = (*Plugin)(nil)
 var _ bot.BotInitHandler = (*Plugin)(nil)
+var _ bot.ShardMigrationHandler = (*Plugin)(nil)
 
 func (p *Plugin) AddCommands() {
 	commands.AddRootCommands(ModerationCommands...)
@@ -53,6 +55,14 @@ func (p *Plugin) BotInit() {
 	pubsub.AddHandler("mod_refresh_mute_override", HandleRefreshMuteOverrides, nil)
 }
 
+func (p *Plugin) GuildMigrated(gs *dstate.GuildState, toThisSlave bool) {
+	if !toThisSlave {
+		return
+	}
+
+	go RefreshMuteOverrides(gs.ID)
+}
+
 func HandleRefreshMuteOverrides(evt *pubsub.Event) {
 	RefreshMuteOverrides(evt.TargetGuildInt)
 }

serverstats/plugin_bot.go

@@ -34,7 +34,6 @@ func (p *Plugin) BotInit() {
 	eventsystem.AddHandler(bot.RedisWrapper(HandleReady), eventsystem.EventReady)
 
 	go UpdateStatsLoop()
-
 }
 
 func (p *Plugin) StopBot(wg *sync.WaitGroup) {

stdcommands/stdcommands.go

@@ -1,6 +1,7 @@
 package stdcommands
 
 import (
+	"github.com/jonas747/yagpdb/bot"
 	"github.com/jonas747/yagpdb/bot/eventsystem"
 	"github.com/jonas747/yagpdb/commands"
 	"github.com/jonas747/yagpdb/common"
@@ -36,9 +37,14 @@ import (
 	"github.com/jonas747/yagpdb/stdcommands/yagstatus"
 )
 
+var (
+	_ bot.BotInitHandler       = (*Plugin)(nil)
+	_ commands.CommandProvider = (*Plugin)(nil)
+)
+
 type Plugin struct{}
 
-func (p *Plugin) InitBot() {
+func (p *Plugin) AddCommands() {
 	commands.AddRootCommands(
 		// Info
 		info.Command,
@@ -76,6 +82,10 @@ func (p *Plugin) InitBot() {
 		memberfetcher.Command,
 		yagstatus.Command,
 	)
+
+}
+
+func (p *Plugin) BotInit() {
 	eventsystem.AddHandler(ping.HandleMessageCreate, eventsystem.EventMessageCreate)
 }
 

streaming/bot.go

@@ -18,6 +18,7 @@ import (
 func KeyCurrentlyStreaming(gID int64) string { return "currently_streaming:" + discordgo.StrID(gID) }
 
 var _ bot.BotInitHandler = (*Plugin)(nil)
+var _ bot.ShardMigrationHandler = (*Plugin)(nil)
 
 func (p *Plugin) BotInit() {
 	eventsystem.AddHandler(bot.ConcurrentEventHandler(bot.RedisWrapper(HandleGuildCreate)), eventsystem.EventGuildCreate)
@@ -26,9 +27,28 @@ func (p *Plugin) BotInit() {
 	pubsub.AddHandler("update_streaming", HandleUpdateStreaming, nil)
 }
 
+func (p *Plugin) GuildMigrated(gs *dstate.GuildState, toThisSlave bool) {
+	if !toThisSlave {
+		return
+	}
+
+	go CheckGuildFull(gs)
+}
+
 // YAGPDB event
 func HandleUpdateStreaming(event *pubsub.Event) {
-	log.Info("Received the streaming event boi ", event.TargetGuild)
+	log.Info("Received update streaming event ", event.TargetGuild)
+
+	gs := bot.State.Guild(true, event.TargetGuildInt)
+	if gs == nil {
+		return
+	}
+
+	CheckGuildFull(gs)
+}
+
+func CheckGuildFull(gs *dstate.GuildState) {
+	log.Info("Streaming Checking full guild: ", gs.ID)
 
 	client, err := common.RedisPool.Get()
 	if err != nil {
@@ -37,11 +57,6 @@ func HandleUpdateStreaming(event *pubsub.Event) {
 	}
 	defer common.RedisPool.Put(client)
 
-	gs := bot.State.Guild(true, event.TargetGuildInt)
-	if gs == nil {
-		return
-	}
-
 	config, err := GetConfig(client, gs.ID)
 	if err != nil {
 		log.WithError(err).WithField("guild", gs.ID).Error("Failed retrieving streaming config")