Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -2,3 +2,4 @@
/bin/
/libexec/
/.vendor/
.idea/
8 changes: 8 additions & 0 deletions go/base/context.go
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@ type MigrationContext struct {
ThrottleAdditionalFlagFile string
throttleQuery string
throttleHTTP string
IgnoreHTTPErrors bool
ThrottleCommandedByUser int64
HibernateUntil int64
maxLoad LoadMap
Expand Down Expand Up @@ -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()
Expand Down
2 changes: 2 additions & 0 deletions go/cmd/gh-ost/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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 {
Expand Down
26 changes: 22 additions & 4 deletions go/logic/throttler.go
Original file line number Diff line number Diff line change
Expand Up @@ -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",
}
)

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -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)
}
}
Expand Down