feat: refactor Fetcher interface used for downloading migrations (#8728)
* feat: refactor Fetcher interface used for downloading migrations * feat: add RetryFetcher for migration downloads * feat: 3 retries for each HTTP migration download (cherry picked from commit b1ffc870d5f2fe167458918311b06ecf795d3422)
Adin Schmahmann committed
Feb 11, 2022 at 14:57 UTC
bf76ebe6671a04fc1399346695a33fee31f759d3
10 files changed
+83
-44
repo/fsrepo/migrations/fetch.go
+3
-4
@@ -111,15 +111,14 @@ func FetchBinary(ctx context.Context, fetcher Fetcher, dist, ver, binName, out s
111
}
112
defer arcFile.Close()
113
114
- // Open connection to download archive from ipfs path
115
- rc, err := fetcher.Fetch(ctx, arcDistPath)
114
+ // Open connection to download archive from ipfs path and write to file
115
+ arcBytes, err := fetcher.Fetch(ctx, arcDistPath)
116
if err != nil {
117
return "", err
118
}
119
- defer rc.Close()
119
120
// Write download data
122
- _, err = io.Copy(arcFile, rc)
121
+ _, err = io.Copy(arcFile, bytes.NewReader(arcBytes))
122
if err != nil {
123
return "", err
124
}
repo/fsrepo/migrations/fetch_test.go
+4
-10
@@ -2,10 +2,10 @@ package migrations
2
3
import (
4
"bufio"
5
+ "bytes"
6
"context"
7
"fmt"
8
"io"
8
- "io/ioutil"
9
"net/http"
10
"net/http/httptest"
11
"os"
@@ -96,14 +96,13 @@ func TestHttpFetch(t *testing.T) {
96
97
fetcher := NewHttpFetcher("", ts.URL, "", 0)
98
99
- rc, err := fetcher.Fetch(ctx, "/versions")
99
+ out, err := fetcher.Fetch(ctx, "/versions")
100
if err != nil {
101
t.Fatal(err)
102
}
103
- defer rc.Close()
103
104
var lines []string
106
- scan := bufio.NewScanner(rc)
105
+ scan := bufio.NewScanner(bytes.NewReader(out))
106
for scan.Scan() {
107
lines = append(lines, scan.Text())
108
}
@@ -232,16 +231,11 @@ func TestMultiFetcher(t *testing.T) {
231
232
mf := NewMultiFetcher(badFetcher, fetcher)
233
235
- rc, err := mf.Fetch(ctx, "/versions")
234
+ vers, err := mf.Fetch(ctx, "/versions")
235
if err != nil {
236
t.Fatal(err)
237
}
239
- defer rc.Close()
238
241
- vers, err := ioutil.ReadAll(rc)
242
- if err != nil {
243
- t.Fatal("could not read versions:", err)
244
- }
239
if len(vers) < 45 {
240
fmt.Println("unexpected more data")
241
}
repo/fsrepo/migrations/fetcher.go
+4
-6
@@ -21,8 +21,7 @@ const (
21
22
type Fetcher interface {
23
// Fetch attempts to fetch the file at the given ipfs path.
24
- // Returns io.ReadCloser on success, which caller must close.
25
- Fetch(ctx context.Context, filePath string) (io.ReadCloser, error)
24
+ Fetch(ctx context.Context, filePath string) ([]byte, error)
25
// Close performs any cleanup after the fetcher is not longer needed.
26
Close() error
27
}
@@ -50,13 +49,12 @@ func NewMultiFetcher(f ...Fetcher) *MultiFetcher {
49
}
50
51
// Fetch attempts to fetch the file at each of its fetchers until one succeeds.
53
-// Returns io.ReadCloser on success, which caller must close.
54
-func (f *MultiFetcher) Fetch(ctx context.Context, ipfsPath string) (io.ReadCloser, error) {
52
+func (f *MultiFetcher) Fetch(ctx context.Context, ipfsPath string) ([]byte, error) {
53
var errs error
54
for _, fetcher := range f.fetchers {
57
- rc, err := fetcher.Fetch(ctx, ipfsPath)
55
+ out, err := fetcher.Fetch(ctx, ipfsPath)
56
if err == nil {
59
- return rc, nil
57
+ return out, nil
58
}
59
fmt.Printf("Error fetching: %s\n", err.Error())
60
errs = multierror.Append(errs, err)
repo/fsrepo/migrations/httpfetcher.go
+9
-5
@@ -60,9 +60,8 @@ func NewHttpFetcher(distPath, gateway, userAgent string, fetchLimit int64) *Http
60
}
61
62
// Fetch attempts to fetch the file at the given path, from the distribution
63
-// site configured for this HttpFetcher. Returns io.ReadCloser on success,
64
-// which caller must close.
65
-func (f *HttpFetcher) Fetch(ctx context.Context, filePath string) (io.ReadCloser, error) {
63
+// site configured for this HttpFetcher.
64
+func (f *HttpFetcher) Fetch(ctx context.Context, filePath string) ([]byte, error) {
65
gwURL := f.gateway + path.Join(f.distPath, filePath)
66
fmt.Printf("Fetching with HTTP: %q\n", gwURL)
67
@@ -89,10 +88,15 @@ func (f *HttpFetcher) Fetch(ctx context.Context, filePath string) (io.ReadCloser
88
return nil, fmt.Errorf("GET %s error: %s: %s", gwURL, resp.Status, string(mes))
89
}
90
91
+ var rc io.ReadCloser
92
if f.limit != 0 {
93
- return NewLimitReadCloser(resp.Body, f.limit), nil
93
+ rc = NewLimitReadCloser(resp.Body, f.limit)
94
+ } else {
95
+ rc = resp.Body
96
}
95
- return resp.Body, nil
97
+ defer rc.Close()
98
+
99
+ return ioutil.ReadAll(rc)
100
}
101
102
func (f *HttpFetcher) Close() error {
repo/fsrepo/migrations/ipfsfetcher/ipfsfetcher.go
+11
-5
@@ -52,6 +52,8 @@ type IpfsFetcher struct {
52
addrInfo peer.AddrInfo
53
}
54
55
+var _ migrations.Fetcher = (*IpfsFetcher)(nil)
56
+
57
// NewIpfsFetcher creates a new IpfsFetcher
58
//
59
// Specifying "" for distPath sets the default IPNS path.
@@ -85,9 +87,8 @@ func NewIpfsFetcher(distPath string, fetchLimit int64, repoRoot *string) *IpfsFe
87
}
88
89
// Fetch attempts to fetch the file at the given path, from the distribution
88
-// site configured for this HttpFetcher. Returns io.ReadCloser on success,
89
-// which caller must close.
90
-func (f *IpfsFetcher) Fetch(ctx context.Context, filePath string) (io.ReadCloser, error) {
90
+// site configured for this HttpFetcher.
91
+func (f *IpfsFetcher) Fetch(ctx context.Context, filePath string) ([]byte, error) {
92
// Initialize and start IPFS node on first call to Fetch, since the fetcher
93
// may be created by not used.
94
f.openOnce.Do(func() {
@@ -123,10 +124,15 @@ func (f *IpfsFetcher) Fetch(ctx context.Context, filePath string) (io.ReadCloser
124
return nil, fmt.Errorf("%q is not a file", filePath)
125
}
126
127
+ var rc io.ReadCloser
128
if f.limit != 0 {
127
- return migrations.NewLimitReadCloser(fileNode, f.limit), nil
129
+ rc = migrations.NewLimitReadCloser(fileNode, f.limit)
130
+ } else {
131
+ rc = fileNode
132
}
129
- return fileNode, nil
133
+ defer rc.Close()
134
+
135
+ return ioutil.ReadAll(rc)
136
}
137
138
func (f *IpfsFetcher) Close() error {
repo/fsrepo/migrations/ipfsfetcher/ipfsfetcher_test.go
+4
-5
@@ -2,6 +2,7 @@ package ipfsfetcher
2
3
import (
4
"bufio"
5
+ "bytes"
6
"context"
7
"fmt"
8
"os"
@@ -28,14 +29,13 @@ func TestIpfsFetcher(t *testing.T) {
29
fetcher := NewIpfsFetcher("", 0, nil)
30
defer fetcher.Close()
31
31
- rc, err := fetcher.Fetch(ctx, "go-ipfs/versions")
32
+ out, err := fetcher.Fetch(ctx, "go-ipfs/versions")
33
if err != nil {
34
t.Fatal(err)
35
}
35
- defer rc.Close()
36
37
var lines []string
38
- scan := bufio.NewScanner(rc)
38
+ scan := bufio.NewScanner(bytes.NewReader(out))
39
for scan.Scan() {
40
lines = append(lines, scan.Text())
41
}
@@ -52,8 +52,7 @@ func TestIpfsFetcher(t *testing.T) {
52
}
53
54
// Check not found
55
- _, err = fetcher.Fetch(ctx, "/no_such_file")
56
- if err == nil {
55
+ if _, err = fetcher.Fetch(ctx, "/no_such_file"); err == nil {
56
t.Fatal("expected error 404")
57
}
58
repo/fsrepo/migrations/migrations.go
+3
-2
@@ -155,13 +155,14 @@ func ReadMigrationConfig(repoRoot string) (*config.Migration, error) {
155
// downloadSources,
156
func GetMigrationFetcher(downloadSources []string, distPath string, newIpfsFetcher func(string) Fetcher) (Fetcher, error) {
157
const httpUserAgent = "go-ipfs"
158
+ const numTriesPerHTTP = 3
159
160
var fetchers []Fetcher
161
for _, src := range downloadSources {
162
src := strings.TrimSpace(src)
163
switch src {
164
case "HTTPS", "https", "HTTP", "http":
164
- fetchers = append(fetchers, NewHttpFetcher(distPath, "", httpUserAgent, 0))
165
+ fetchers = append(fetchers, &RetryFetcher{NewHttpFetcher(distPath, "", httpUserAgent, 0), numTriesPerHTTP})
166
case "IPFS", "ipfs":
167
if newIpfsFetcher != nil {
168
fetchers = append(fetchers, newIpfsFetcher(distPath))
@@ -178,7 +179,7 @@ func GetMigrationFetcher(downloadSources []string, distPath string, newIpfsFetch
179
default:
180
return nil, errors.New("bad gateway address: url scheme must be http or https")
181
}
181
- fetchers = append(fetchers, NewHttpFetcher(distPath, u.String(), httpUserAgent, 0))
182
+ fetchers = append(fetchers, &RetryFetcher{NewHttpFetcher(distPath, u.String(), httpUserAgent, 0), numTriesPerHTTP})
183
case "":
184
// Ignore empty string
185
}
repo/fsrepo/migrations/migrations_test.go
+9
-4
@@ -3,7 +3,6 @@ package migrations
3
import (
4
"context"
5
"fmt"
6
- "io"
6
"log"
7
"os"
8
"path/filepath"
@@ -290,7 +289,9 @@ func TestReadMigrationConfig(t *testing.T) {
289
290
type mockIpfsFetcher struct{}
291
293
-func (m *mockIpfsFetcher) Fetch(ctx context.Context, filePath string) (io.ReadCloser, error) {
292
+var _ Fetcher = (*mockIpfsFetcher)(nil)
293
+
294
+func (m *mockIpfsFetcher) Fetch(ctx context.Context, filePath string) ([]byte, error) {
295
return nil, nil
296
}
297
@@ -323,7 +324,9 @@ func TestGetMigrationFetcher(t *testing.T) {
324
if err != nil {
325
t.Fatal(err)
326
}
326
- if _, ok := f.(*HttpFetcher); !ok {
327
+ if rf, ok := f.(*RetryFetcher); !ok {
328
+ t.Fatal("expected RetryFetcher")
329
+ } else if _, ok := rf.Fetcher.(*HttpFetcher); !ok {
330
t.Fatal("expected HttpFetcher")
331
}
332
@@ -341,7 +344,9 @@ func TestGetMigrationFetcher(t *testing.T) {
344
if err != nil {
345
t.Fatal(err)
346
}
344
- if _, ok := f.(*HttpFetcher); !ok {
347
+ if rf, ok := f.(*RetryFetcher); !ok {
348
+ t.Fatal("expected RetryFetcher")
349
+ } else if _, ok := rf.Fetcher.(*HttpFetcher); !ok {
350
t.Fatal("expected HttpFetcher")
351
}
352
repo/fsrepo/migrations/retryfetcher.go
new
+33
@@ -0,0 +1,33 @@
1
+package migrations
2
+
3
+import (
4
+ "context"
5
+ "fmt"
6
+)
7
+
8
+type RetryFetcher struct {
9
+ Fetcher
10
+ MaxTries int
11
+}
12
+
13
+var _ Fetcher = (*RetryFetcher)(nil)
14
+
15
+func (r *RetryFetcher) Fetch(ctx context.Context, filePath string) ([]byte, error) {
16
+ var lastErr error
17
+ for i := 0; i < r.MaxTries; i++ {
18
+ out, err := r.Fetcher.Fetch(ctx, filePath)
19
+ if err == nil {
20
+ return out, nil
21
+ }
22
+
23
+ if ctx.Err() != nil {
24
+ return nil, ctx.Err()
25
+ }
26
+ lastErr = err
27
+ }
28
+ return nil, fmt.Errorf("exceeded number of retries. last error was %w", lastErr)
29
+}
30
+
31
+func (r *RetryFetcher) Close() error {
32
+ return r.Fetcher.Close()
33
+}
repo/fsrepo/migrations/versions.go
+3
-3
@@ -2,6 +2,7 @@ package migrations
2
3
import (
4
"bufio"
5
+ "bytes"
6
"context"
7
"errors"
8
"fmt"
@@ -39,16 +40,15 @@ func LatestDistVersion(ctx context.Context, fetcher Fetcher, dist string, stable
40
// available on the distriburion site. List is in ascending order, unless
41
// sortDesc is true.
42
func DistVersions(ctx context.Context, fetcher Fetcher, dist string, sortDesc bool) ([]string, error) {
42
- rc, err := fetcher.Fetch(ctx, path.Join(dist, distVersions))
43
+ versionBytes, err := fetcher.Fetch(ctx, path.Join(dist, distVersions))
44
if err != nil {
45
return nil, err
46
}
46
- defer rc.Close()
47
48
prefix := "v"
49
var vers []semver.Version
50
51
- scan := bufio.NewScanner(rc)
51
+ scan := bufio.NewScanner(bytes.NewReader(versionBytes))
52
for scan.Scan() {
53
ver, err := semver.Make(strings.TrimLeft(scan.Text(), prefix))
54
if err != nil {