diff --git a/.gitignore b/.gitignore index 63f0df9..605546d 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ /bin/ /libexec/ /.vendor/ +.idea/ diff --git a/go/base/context.go b/go/base/context.go index 4220422..15cae60 100644 --- a/go/base/context.go +++ b/go/base/context.go @@ -120,6 +120,7 @@ type MigrationContext struct { ThrottleAdditionalFlagFile string throttleQuery string throttleHTTP string + IgnoreHTTPErrors bool ThrottleCommandedByUser int64 HibernateUntil int64 maxLoad LoadMap @@ -575,6 +576,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() diff --git a/go/cmd/gh-ost/main.go b/go/cmd/gh-ost/main.go index 9051465..e7c05ba 100644 --- a/go/cmd/gh-ost/main.go +++ b/go/cmd/gh-ost/main.go @@ -107,6 +107,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") @@ -260,6 +261,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 { diff --git a/go/logic/throttler.go b/go/logic/throttler.go index 6f9f4bb..2cf0d97 100644 --- a/go/logic/throttler.go +++ b/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) } }