Merge pull request #833 from github/jfudally-throttle-fix

Throttle on HTTP throttle error

.gitignore

@@ -2,3 +2,4 @@
 /bin/
 /libexec/
 /.vendor/
+.idea/

go/base/context.go

@@ -119,6 +119,7 @@ type MigrationContext struct {
 	ThrottleAdditionalFlagFile          string
 	throttleQuery                       string
 	throttleHTTP                        string
+	IgnoreHTTPErrors                    bool
 	ThrottleCommandedByUser             int64
 	HibernateUntil                      int64
 	maxLoad                             LoadMap
@@ -574,6 +575,13 @@ func (this *MigrationContext) SetThrottleHTTP(throttleHTTP string) {
 	this.throttleHTTP = throttleHTTP
 }
 
+func (this *MigrationContext) SetIgnoreHTTPErrors(ignoreHTTPErrors bool) {
+	this.throttleHTTPMutex.Lock()
+	defer this.throttleHTTPMutex.Unlock()
+
+	this.IgnoreHTTPErrors = ignoreHTTPErrors
+}
+
 func (this *MigrationContext) GetMaxLoad() LoadMap {
 	this.throttleMutex.Lock()
 	defer this.throttleMutex.Unlock()

go/cmd/gh-ost/main.go

@@ -106,6 +106,7 @@ func main() {
 	throttleControlReplicas := flag.String("throttle-control-replicas", "", "List of replicas on which to check for lag; comma delimited. Example: myhost1.com:3306,myhost2.com,myhost3.com:3307")
 	throttleQuery := flag.String("throttle-query", "", "when given, issued (every second) to check if operation should throttle. Expecting to return zero for no-throttle, >0 for throttle. Query is issued on the migrated server. Make sure this query is lightweight")
 	throttleHTTP := flag.String("throttle-http", "", "when given, gh-ost checks given URL via HEAD request; any response code other than 200 (OK) causes throttling; make sure it has low latency response")
+	ignoreHTTPErrors := flag.Bool("ignore-http-errors", false, "ignore HTTP connection errors during throttle check")
 	heartbeatIntervalMillis := flag.Int64("heartbeat-interval-millis", 100, "how frequently would gh-ost inject a heartbeat value")
 	flag.StringVar(&migrationContext.ThrottleFlagFile, "throttle-flag-file", "", "operation pauses when this file exists; hint: use a file that is specific to the table being altered")
 	flag.StringVar(&migrationContext.ThrottleAdditionalFlagFile, "throttle-additional-flag-file", "/tmp/gh-ost.throttle", "operation pauses when this file exists; hint: keep default, use for throttling multiple gh-ost operations")
@@ -259,6 +260,7 @@ func main() {
 	migrationContext.SetMaxLagMillisecondsThrottleThreshold(*maxLagMillis)
 	migrationContext.SetThrottleQuery(*throttleQuery)
 	migrationContext.SetThrottleHTTP(*throttleHTTP)
+	migrationContext.SetIgnoreHTTPErrors(*ignoreHTTPErrors)
 	migrationContext.SetDefaultNumRetries(*defaultRetries)
 	migrationContext.ApplyCredentials()
 	if err := migrationContext.SetupTLS(); err != nil {

go/logic/throttler.go

@@ -19,20 +19,22 @@ import (
 )
 
 var (
-	httpStatusMessages map[int]string = map[int]string{
+	httpStatusMessages = map[int]string{
 		200: "OK",
 		404: "Not found",
 		417: "Expectation failed",
 		429: "Too many requests",
 		500: "Internal server error",
+		-1:  "Connection error",
 	}
 	// See https://github.com/github/freno/blob/master/doc/http.md
-	httpStatusFrenoMessages map[int]string = map[int]string{
+	httpStatusFrenoMessages = map[int]string{
 		200: "OK",
 		404: "freno: unknown metric",
 		417: "freno: access forbidden",
 		429: "freno: threshold exceeded",
 		500: "freno: internal error",
+		-1:  "freno: connection error",
 	}
 )
 
@@ -84,6 +86,7 @@ func (this *Throttler) shouldThrottle() (result bool, reason string, reasonHint
 	if statusCode != 0 && statusCode != http.StatusOK {
 		return true, this.throttleHttpMessage(int(statusCode)), base.NoThrottleReasonHint
 	}
+
 	// Replication lag throttle
 	maxLagMillisecondsThrottleThreshold := atomic.LoadInt64(&this.migrationContext.MaxLagMillisecondsThrottleThreshold)
 	lag := atomic.LoadInt64(&this.migrationContext.CurrentLag)
@@ -288,7 +291,14 @@ func (this *Throttler) collectThrottleHTTPStatus(firstThrottlingCollected chan<-
 		return false, nil
 	}
 
-	collectFunc()
+	_, err := collectFunc()
+	if err != nil {
+		// If not told to ignore errors, we'll throttle on HTTP connection issues
+		if !this.migrationContext.IgnoreHTTPErrors {
+			atomic.StoreInt64(&this.migrationContext.ThrottleHTTPStatusCode, int64(-1))
+		}
+	}
+
 	firstThrottlingCollected <- true
 
 	ticker := time.Tick(100 * time.Millisecond)
@@ -297,7 +307,15 @@ func (this *Throttler) collectThrottleHTTPStatus(firstThrottlingCollected chan<-
 			return
 		}
 
-		if sleep, _ := collectFunc(); sleep {
+		sleep, err := collectFunc()
+		if err != nil {
+			// If not told to ignore errors, we'll throttle on HTTP connection issues
+			if !this.migrationContext.IgnoreHTTPErrors {
+				atomic.StoreInt64(&this.migrationContext.ThrottleHTTPStatusCode, int64(-1))
+			}
+		}
+
+		if sleep {
 			time.Sleep(1 * time.Second)
 		}
 	}