-
Notifications
You must be signed in to change notification settings - Fork 433
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Add s3 download by range * Add config add devel * Full remake * Small fixes * Retrigger * Fix exp backoff for s3reader * Simplified debug log for s3Reader * Extract s3reader * go fmt changes * Removed debug line * Reuse last Body * Fix config, add test for mysql with ranges * Small rename * Add comment, fix test, fix error msg * Rename bucket * Add devel, fix prefix for mysql test * Fix docs
- Loading branch information
Showing
10 changed files
with
276 additions
and
23 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
37 changes: 37 additions & 0 deletions
37
docker/mysql_tests/scripts/base_tests/full_xtrabackup_test_with_ranges.sh
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| #!/bin/sh | ||
| set -e -x | ||
|
|
||
| . /usr/local/export_common.sh | ||
|
|
||
| export WALG_LOG_LEVEL=DEVEL | ||
| export WALE_S3_PREFIX=s3://mysqlfullxtrabackupwithrangesbucket | ||
| export WALG_S3_RANGE_BATCH_ENABLED=true | ||
|
|
||
|
|
||
| mysqld --initialize --init-file=/etc/mysql/init.sql | ||
|
|
||
| service mysql start | ||
|
|
||
| sysbench --table-size=10 prepare | ||
|
|
||
| sysbench --time=5 run | ||
|
|
||
| mysql -e 'FLUSH LOGS' | ||
|
|
||
| mysqldump sbtest > /tmp/dump_before_backup | ||
|
|
||
| wal-g backup-push | ||
|
|
||
| mysql_kill_and_clean_data | ||
|
|
||
| wal-g backup-fetch LATEST | ||
|
|
||
| chown -R mysql:mysql $MYSQLDATA | ||
|
|
||
| service mysql start || (cat /var/log/mysql/error.log && false) | ||
|
|
||
| mysql_set_gtid_purged | ||
|
|
||
| mysqldump sbtest > /tmp/dump_after_restore | ||
|
|
||
| diff /tmp/dump_before_backup /tmp/dump_after_restore |
3 changes: 2 additions & 1 deletion
3
docker/pg_tests/scripts/configs/several_delta_backups_reverse_test_config.json
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,4 @@ | ||
| "WALE_S3_PREFIX": "s3://compressionreversebucket", | ||
| "WALG_DELTA_MAX_STEPS": "6", | ||
| "WALG_USE_WAL_DELTA": "true" | ||
| "WALG_USE_WAL_DELTA": "true", | ||
| "WALG_LOG_LEVEL": "DEVEL" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,165 @@ | ||
| package s3 | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "hash/fnv" | ||
| "io" | ||
| "math" | ||
| "math/rand" | ||
| "strconv" | ||
| "time" | ||
|
|
||
| "github.com/aws/aws-sdk-go/aws" | ||
| "github.com/aws/aws-sdk-go/service/s3" | ||
| "github.com/pkg/errors" | ||
| "github.com/wal-g/tracelog" | ||
| ) | ||
|
|
||
| var ( | ||
| DebugLogBufferCounter = 0 | ||
| ) | ||
|
|
||
| type s3Reader struct { | ||
| lastBody io.ReadCloser | ||
| folder *Folder | ||
| maxRetries int | ||
| retryNum int | ||
| objectPath string | ||
| storageCursor int64 | ||
| maxRetryDelay time.Duration | ||
| minRetryDelay time.Duration | ||
| reconnectId int | ||
| logDebugId string // hash from filename and logDebugId - unique logDebugId used only for debug | ||
| } | ||
|
|
||
| func (reader *s3Reader) getObjectRange(from, to int64) (*s3.GetObjectOutput, error) { | ||
| bytesRange := fmt.Sprintf("bytes=%d-", from) | ||
| if to != 0 { | ||
| bytesRange += strconv.Itoa(int(to)) | ||
| } | ||
| input := &s3.GetObjectInput{ | ||
| Bucket: reader.folder.Bucket, | ||
| Key: aws.String(reader.objectPath), | ||
| Range: aws.String(bytesRange), | ||
| } | ||
| reader.debugLog("GetObject with range %s", bytesRange) | ||
| return reader.folder.S3API.GetObject(input) | ||
| } | ||
|
|
||
| func (reader *s3Reader) Read(p []byte) (n int, err error) { | ||
| reader.debugLog("Read to buffer [%d] bytes", len(p)) | ||
| reconnect := false | ||
| if reader.lastBody == nil { // initial connect, if lastBody wasn't provided | ||
| reconnect = true | ||
| } | ||
| for { | ||
| if reconnect { | ||
| reconnect = false | ||
| connErr := reader.reconnect() | ||
| if connErr != nil { | ||
| reader.debugLog("reconnect failed %s", connErr) | ||
| return 0, connErr | ||
| } | ||
| } | ||
|
|
||
| n, err = reader.lastBody.Read(p) | ||
| reader.debugLog("read %d, err %s", n, err) | ||
| if err != nil && err != io.EOF { | ||
| reconnect = true | ||
| continue | ||
| } | ||
| reader.storageCursor += int64(n) | ||
| reader.debugLog("success read") | ||
| return n, err | ||
| } | ||
| } | ||
| func (reader *s3Reader) getDebugLogLine(format string, v ...interface{}) string { | ||
| prefix := fmt.Sprintf("s3Reader [%s] ", reader.logDebugId) | ||
| message := fmt.Sprintf(format, v...) | ||
| return prefix + message | ||
| } | ||
|
|
||
| func (reader *s3Reader) debugLog(format string, v ...interface{}) { | ||
| tracelog.DebugLogger.Print(reader.getDebugLogLine(format, v...)) | ||
| } | ||
|
|
||
| func (reader *s3Reader) reconnect() error { | ||
| failed := 0 | ||
|
|
||
| for { | ||
| reader.reconnectId++ | ||
| object, err := reader.getObjectRange(reader.storageCursor, 0) | ||
| if err != nil { | ||
| failed += 1 | ||
| reader.debugLog("reconnect failed [%d/%d]: %s", failed, reader.maxRetries, err) | ||
| if failed >= reader.maxRetries { | ||
| return errors.Wrap(err, reader.getDebugLogLine("Too much reconnecting retries")) | ||
| } | ||
| sleepTime := reader.getIncrSleep(failed) | ||
| reader.debugLog("sleep: %s", sleepTime) | ||
| time.Sleep(sleepTime) | ||
| continue | ||
| } | ||
| failed = 0 | ||
| if reader.lastBody != nil { | ||
| err = reader.lastBody.Close() | ||
| if err != nil { | ||
| msg := reader.getDebugLogLine("We have problems with closing previous connection") | ||
| tracelog.DebugLogger.Print(msg) | ||
| return errors.Wrap(err, msg) | ||
| } | ||
| } | ||
| reader.lastBody = object.Body | ||
| reader.debugLog("reconnect #%d succeeded", reader.reconnectId) | ||
| break | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // THIS COde stolen from s3 lib, from vendor/github.com/aws/aws-sdk-go/aws/client/default_retryer.go | ||
| // func (d DefaultRetryer) RetryRules( .. ) time.Duration | ||
| // this calculate sleep duration (jitter and exponential backoff) | ||
| func (reader *s3Reader) getIncrSleep(retryCount int) time.Duration { | ||
| minDelay := reader.minRetryDelay | ||
| maxDelay := reader.maxRetryDelay | ||
| var delay time.Duration | ||
|
|
||
| actualRetryCount := int(math.Log2(float64(minDelay))) + 1 | ||
| if actualRetryCount < 63-retryCount { | ||
| delay = time.Duration(1<<uint64(retryCount)) * getJitterDelay(minDelay) | ||
| if delay > maxDelay { | ||
| delay = getJitterDelay(maxDelay / 2) | ||
| } | ||
| } else { | ||
| delay = getJitterDelay(maxDelay / 2) | ||
| } | ||
| return delay | ||
| } | ||
|
|
||
| func (reader *s3Reader) Close() (err error) { | ||
| return reader.lastBody.Close() | ||
| } | ||
|
|
||
| func NewS3Reader(body io.ReadCloser, objectPath string, retriesCount int, folder *Folder, | ||
| minRetryDelay, maxRetryDelay time.Duration) *s3Reader { | ||
|
|
||
| DebugLogBufferCounter++ | ||
| reader := &s3Reader{lastBody: body, objectPath: objectPath, maxRetries: retriesCount, logDebugId: getHash(objectPath, DebugLogBufferCounter), | ||
| folder: folder, minRetryDelay: minRetryDelay, maxRetryDelay: maxRetryDelay} | ||
|
|
||
| reader.debugLog("Init s3reader path %s", objectPath) | ||
| return reader | ||
| } | ||
|
|
||
| func getHash(objectPath string, id int) string { | ||
| hash := fnv.New32a() | ||
| _, err := hash.Write([]byte(objectPath)) | ||
| tracelog.ErrorLogger.FatalfOnError("Fatal, can't write buffer to hash %v", err) | ||
|
|
||
| return fmt.Sprintf("%x_%d", hash.Sum32(), id) | ||
| } | ||
|
|
||
| // getJitterDelay returns a jittered delay for retry | ||
| func getJitterDelay(duration time.Duration) time.Duration { | ||
| return time.Duration(rand.Int63n(int64(duration)) + int64(duration)) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -167,4 +167,4 @@ func getEndpointPort(settings map[string]string) string { | |
| return port | ||
| } | ||
| return DefaultPort | ||
| } | ||
| } | ||
Oops, something went wrong.