Crash service on certain errors, so kubernetes can restart. Several fixes

Author: Gleez Technologies

cmd/mysql-manticore/main.go

@@ -146,20 +146,21 @@ func run() (err error) {
 	select {
 	case n := <-sc:
 		log.Infof("received signal %v, exiting", n)
-		// case err = <-r.FatalErrC:
-		// 	if errors.Cause(err) == river.ErrRebuildAndExitFlagSet {
-		// 		log.Info(err.Error())
-		// 		err = nil
-		// 	}
+	case err = <-r.FatalErrC:
+		// if errors.Cause(err) == river.ErrRebuildAndExitFlagSet {
+		// 	log.Info(err.Error())
+		// 	err = nil
+		// }
+		log.Infof("received error [%v], exiting", err.Error())
 	}
 
 	rootSup.Stop()
 	return err
 }
 
-// k8s run master loop will check for master once per 30 seconds.
+// k8s run master loop will check for master once per 20 seconds.
 func runMasterLoop() {
-	ticker := time.NewTicker(30 * time.Second)
+	ticker := time.NewTicker(10 * time.Second)
 	for {
 		select {
 		case <-done:

river/river.go

@@ -181,39 +181,38 @@ func (r *River) run() error {
 
 	r.sup.ServeBackground()
 
-	r.sphinxService.RequestStartNotification()
+	if r.sphinxToken == nil {
+		r.sphinxService.RequestStartNotification()
 
-	t := r.sup.Add(r.sphinxService)
-	r.sphinxToken = &t
+		t := r.sup.Add(r.sphinxService)
+		r.sphinxToken = &t
 
-	r.sphinxService.WaitUntilStarted()
+		r.sphinxService.WaitUntilStarted()
+	}
 
 	b := &backoff.Backoff{
 		Min:    1 * time.Second,
-		Max:    20 * time.Minute,
+		Max:    10 * time.Minute,
 		Factor: 2,
 		Jitter: true,
 	}
 	defer b.Reset()
 
-	// get master state - wait until get state or timeout
-	for {
-		time.Sleep(b.Duration())
-
-		err = r.sphinxService.LoadSyncState(r.master.syncState())
-		if err != nil {
-			r.l.Errorf("one or more manticore backends are not up to date: %v", err)
-		}
+	// // get master state - wait until get state or timeout
+	// for {
+	// 	time.Sleep(b.Duration())
 
-		if err == nil {
-			b.Reset()
-			r.l.Infof("Connected to manticore backend")
-			break
-		}
-	}
+	// 	err = r.sphinxService.LoadSyncState(r.master.syncState())
+	// 	if err == nil {
+	// 		b.Reset()
+	// 		r.l.Infof("Connected to manticore backend")
+	// 		break
+	// 	}
+	// }
 
+	err = r.sphinxService.LoadSyncState(r.master.syncState())
 	if err != nil {
-		// r.l.Errorf("one or more manticore backends are not up to date: %v", err)
+		r.l.Errorf("one or more manticore backends are not up to date: %v", err)
 		return errors.Trace(err)
 	}
 

river/sphinx_service.go

@@ -6,6 +6,7 @@ import (
 	"github.com/juju/errors"
 	"github.com/sandeepone/mysql-manticore/sphinx"
 	"github.com/siddontang/go/sync2"
+
 	"gopkg.in/birkirb/loggers.v1"
 	"gopkg.in/birkirb/loggers.v1/log"
 )
@@ -125,7 +126,13 @@ func (s *SphinxService) SaveSyncState() (err error) {
 	if s.sph == nil {
 		return errors.Trace(errSphinxDisconnected)
 	}
-	return errors.Trace(sphinx.SaveSyncState(s.sph, master.syncState()))
+
+	err = sphinx.SaveSyncState(s.sph, master.syncState())
+	if err != nil {
+		s.riverInstance.FatalErrC <- errors.Annotatef(err, "could not connect to manticore")
+	}
+
+	return errors.Trace(err)
 }
 
 // IndexIsReady ...
@@ -138,22 +145,6 @@ func (s *SphinxService) IndexIsReady(index string, parts uint16) (bool, error) {
 	return sphinx.IndexIsReady(s.sph, index, parts)
 }
 
-// ReloadRtIndex ...
-// func (s *SphinxService) ReloadRtIndex(build indexGroupBuild) error {
-// 	s.sphm.Lock()
-// 	defer s.sphm.Unlock()
-
-// 	if s.sph == nil {
-// 		return errors.Annotatef(errSphinxDisconnected, "error reloading after the build %s", build.id)
-// 	}
-
-// 	if s.riverInstance.balancer != nil {
-// 		return errors.Trace(s.riverInstance.balancer.RollingReloadIndex(s.sph, build))
-// 	}
-
-// 	return nil
-// }
-
 // CheckIndexForOptimize ...
 func (s *SphinxService) CheckIndexForOptimize(index string, parts uint16) error {
 	s.sphm.Lock()

river/status.go

@@ -214,7 +214,7 @@ func (s *stat) run() (err error) {
 
 	getStatusInfo = s.getStatusInfo
 
-	// signal Kubernetes the server is healthy & ready to receive traffic
+	// signal Kubernetes the server is healthy
 	atomic.StoreInt32(&healthy, 1)
 
 	return s.srv.Serve(s.l)
@@ -252,8 +252,8 @@ func handleReadyz(s *stat) http.HandlerFunc {
 		d := time.Now().Sub(s.startedAt)
 
 		// allows to take leadership or rolling update.
-		// wait for 2 mins and inform kubernetes if unsuccessful
-		if !s.r.isRunning && d.Seconds() < 65 {
+		// wait for 1 min and inform kubernetes if unsuccessful
+		if !s.r.isRunning && d.Seconds() < 45 {
 			w.WriteHeader(http.StatusNoContent)
 			return
 		}
@@ -304,11 +304,12 @@ func handleStartSync(r *River) http.HandlerFunc {
 
 		err := r.sphinxService.LoadSyncState(r.master.syncState())
 		if err != nil {
-			r.l.Errorf("Status: one or more manticore backends are not up to date: %v", err)
-			w.Write([]byte("manticore backend not up to date\n"))
+			r.l.Errorf("Status: failed to reset GTID after successful restart: %s", errors.ErrorStack(err))
+			w.Write([]byte("failed to reset GTID\n"))
 			return
 		}
 
+		r.l.Infof("reset GTID after successful restart to: %s", r.master.gtidSet())
 		r.startSyncRoutine()
 		w.WriteHeader(http.StatusNoContent)
 	})

river/sync.go

@@ -48,8 +48,6 @@ type syncState struct {
 	positionEvents []positionEvent
 	flushC         <-chan time.Time
 	flushTimer     *time.Timer
-	// buildModeC       <-chan buildModeMsg
-	// buildModeEnabled bool
 }
 
 type syncActionNeed struct {
@@ -176,7 +174,6 @@ func (s *SyncService) SyncLoop(ctx context.Context) {
 				pos:  r.master.position(),
 			},
 		},
-		// buildModeC: s.switchC,
 	}
 
 	defer r.SaveState()
@@ -224,7 +221,6 @@ func (s *SyncService) SyncLoop(ctx context.Context) {
 			state.firstPositionEventID = syncedPositionEventID
 			state.positionEvents = state.positionEvents[offset:]
 
-			// if !state.buildModeEnabled {
 			bnow := time.Now()
 			if bnow.Sub(lastSavedTime) > 3*time.Second {
 				lastSavedTime = bnow
@@ -237,7 +233,6 @@ func (s *SyncService) SyncLoop(ctx context.Context) {
 					}
 				}
 			}
-			// }
 		}
 	}
 }

sphinx/sphinx.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	set "github.com/deckarep/golang-set"
-	"github.com/jpillora/backoff"
 	"github.com/juju/errors"
 
 	"github.com/sandeepone/mysql-manticore/util"
@@ -131,7 +130,7 @@ func ConnectMany(addrList []string, settings ConnSettings, wg *sync.WaitGroup) (
 			in:         make(chan interface{}),
 			out:        make(chan SphResult),
 			settings:   settings,
-			maxRetries: 10,
+			maxRetries: 5,
 		}
 		if wg != nil {
 			wg.Add(1)
@@ -229,23 +228,14 @@ func (c *SphConn) retry(pause time.Duration, stmt string) (*mysql.Result, error)
 		return nil, errors.Trace(err)
 	}
 
-	b := &backoff.Backoff{
-		Min:    1 * time.Second,
-		Max:    30 * time.Second,
-		Factor: 2,
-		Jitter: true,
-	}
-	defer b.Reset()
-
 	for {
-		log.Infof("[sphinx-query@%s] waiting for %s before a retry", c.addr, b.Duration())
-		time.Sleep(b.Duration())
+		log.Infof("[sphinx-query@%s] waiting for %s before a retry", c.addr, pause)
+		time.Sleep(pause)
 		c.attempts = c.attempts + 1
 
 		err = c.reconnect()
 		if err == nil {
-			b.Reset()
-			c.attempts = 0
+			// c.attempts = 0
 			break
 		}
 
@@ -254,7 +244,7 @@ func (c *SphConn) retry(pause time.Duration, stmt string) (*mysql.Result, error)
 		}
 	}
 
-	b.Reset()
+	c.attempts = 0
 	log.Infof("[sphinx-query@%s] [retry] %s", c.RemoteAddr().String(), stmt)
 	return c.conn.Conn.Execute(stmt)
 }