Review changes
gammazero committed
Feb 26, 2021 at 08:41 UTC
fcbe47bc4be7e08239ee0819cf279a7f596b6e29
13 files changed
+454
-296
cmd/ipfs/daemon.go
+4
-3
@@ -288,9 +288,10 @@ func daemonFunc(req *cmds.Request, re cmds.ResponseEmitter, env cmds.Environment
288
return fmt.Errorf("fs-repo requires migration")
289
}
290
291
- // Fetch migrations using the current distribution
292
- migrations.SetIpfsDistPath(migrations.CurrentIpfsDist)
293
- err = migrations.RunMigration(cctx.Context(), fsrepo.RepoVersion, "")
291
+ fetcher := migrations.NewHttpFetcher()
292
+ // Fetch migrations from current distribution, or location from environ
293
+ fetcher.SetDistPath(migrations.GetDistPathEnv(migrations.CurrentIpfsDist))
294
+ err = migrations.RunMigration(cctx.Context(), fetcher, fsrepo.RepoVersion, "", false)
295
if err != nil {
296
fmt.Println("The migrations of fs-repo failed:")
297
fmt.Printf(" %s\n", err)
repo/fsrepo/migrations/fetch.go
+35
-135
@@ -7,7 +7,6 @@ import (
7
"fmt"
8
"io"
9
"io/ioutil"
10
- "net/http"
10
"os"
11
"os/exec"
12
"path"
@@ -15,68 +14,22 @@ import (
14
"strings"
15
)
16
18
-const (
19
- // Current dirstibution to fetch migrations from
20
- CurrentIpfsDist = "/ipfs/Qme8pJhBidEUXRdpcWLGR2fkG5kdwVnaMh3kabjfP8zz7Y"
21
-
22
- envIpfsDistPath = "IPFS_DIST_PATH"
23
-
24
- // Distribution
25
- gatewayURL = "https://ipfs.io"
26
- ipfsDist = "/ipns/dist.ipfs.io"
27
-
28
- // Maximum download size
29
- fetchSizeLimit = 1024 * 1024 * 512
30
-)
31
-
32
-type limitReadCloser struct {
33
- io.Reader
34
- io.Closer
35
-}
36
-
37
-var ipfsDistPath string
38
-
39
-func init() {
40
- SetIpfsDistPath("")
41
-}
42
-
43
-// SetIpfsDistPath sets the ipfs path to the distribution site. If an empty
44
-// string is given, then the path is set using the IPFS_DIST_PATH environ
45
-// veriable, or the default dns link value if that is not defined.
46
-func SetIpfsDistPath(distPath string) {
47
- if distPath != "" {
48
- ipfsDistPath = distPath
49
- return
50
- }
51
-
52
- if dist := os.Getenv(envIpfsDistPath); dist != "" {
53
- ipfsDistPath = dist
54
- } else {
55
- ipfsDistPath = ipfsDist
56
- }
57
-}
58
-
17
// FetchBinary downloads an archive from the distribution site and unpacks it.
18
//
61
-// The base name of the archive file, inside the distribution directory on
62
-// distribution site, may differ from the distribution name. If it does, then
63
-// specify arcName.
64
-//
19
// The base name of the binary inside the archive may differ from the base
20
// archive name. If it does, then specify binName. For example, the following
21
// is needed because the archive "go-ipfs_v0.7.0_linux-amd64.tar.gz" contains a
22
// binary named "ipfs"
23
//
70
-// FetchBinary(ctx, "go-ipfs", "v0.7.0", "go-ipfs", "ipfs", tmpDir)
24
+// FetchBinary(ctx, fetcher, "go-ipfs", "v0.7.0", "ipfs", tmpDir)
25
//
26
// If out is a directory, then the binary is written to that directory with the
27
// same name it has inside the archive. Otherwise, the binary file is written
28
// to the file named by out.
75
-func FetchBinary(ctx context.Context, dist, ver, arcName, binName, out string) (string, error) {
76
- // If archive base name not specified, then it is same as dist.
77
- if arcName == "" {
78
- arcName = dist
79
- }
29
+func FetchBinary(ctx context.Context, fetcher Fetcher, dist, ver, binName, out string) (string, error) {
30
+ // The archive file name is the base of dist to support possible subdir in
31
+ // dist, for example: "ipfs-repo-migrations/ipfs-11-to-12"
32
+ arcName := path.Base(dist)
33
// If binary base name is not specified, then it is same as archive base name.
34
if binName == "" {
35
binName = arcName
@@ -101,6 +54,18 @@ func FetchBinary(ctx context.Context, dist, ver, arcName, binName, out string) (
54
}
55
// out exists and is a directory, so compose final name
56
out = path.Join(out, binName)
57
+ // Check if the binary already exists in the directory
58
+ fi, err = os.Stat(out)
59
+ if !os.IsNotExist(err) {
60
+ if err != nil {
61
+ return "", err
62
+ }
63
+ return "", &os.PathError{
64
+ Op: "FetchBinary",
65
+ Path: out,
66
+ Err: os.ErrExist,
67
+ }
68
+ }
69
}
70
71
// Create temp directory to store download
@@ -115,11 +80,10 @@ func FetchBinary(ctx context.Context, dist, ver, arcName, binName, out string) (
80
atype = "zip"
81
}
82
118
- arcName = makeArchiveName(arcName, ver, atype)
119
- arcIpfsPath := makeIpfsPath(dist, ver, arcName)
83
+ arcDistPath, arcFullName := makeArchivePath(dist, arcName, ver, atype)
84
85
// Create a file to write the archive data to
122
- arcPath := path.Join(tmpDir, arcName)
86
+ arcPath := path.Join(tmpDir, arcFullName)
87
arcFile, err := os.Create(arcPath)
88
if err != nil {
89
return "", err
@@ -127,7 +91,7 @@ func FetchBinary(ctx context.Context, dist, ver, arcName, binName, out string) (
91
defer arcFile.Close()
92
93
// Open connection to download archive from ipfs path
130
- rc, err := fetch(ctx, arcIpfsPath)
94
+ rc, err := fetcher.Fetch(ctx, arcDistPath)
95
if err != nil {
96
return "", err
97
}
@@ -155,73 +119,6 @@ func FetchBinary(ctx context.Context, dist, ver, arcName, binName, out string) (
119
return out, nil
120
}
121
158
-// fetch attempts to fetch the file at the given ipfs path, first using the
159
-// local ipfs api if available, then using http. Returns io.ReadCloser on
160
-// success, which caller must close.
161
-func fetch(ctx context.Context, ipfsPath string) (io.ReadCloser, error) {
162
- // Try fetching via ipfs daemon
163
- rc, err := ipfsFetch(ctx, ipfsPath)
164
- if err == nil {
165
- // Transferred using local ipfs daemon
166
- return rc, nil
167
- }
168
- // Try fetching via HTTP
169
- return httpFetch(ctx, gatewayURL+ipfsPath)
170
-}
171
-
172
-// ipfsFetch attempts to fetch the file at the given ipfs path using the local
173
-// ipfs api. Returns io.ReadCloser on success, which caller must close.
174
-func ipfsFetch(ctx context.Context, ipfsPath string) (io.ReadCloser, error) {
175
- sh, _, err := ApiShell("")
176
- if err != nil {
177
- return nil, err
178
- }
179
-
180
- resp, err := sh.Request("cat", ipfsPath).Send(ctx)
181
- if err != nil {
182
- return nil, err
183
- }
184
- if resp.Error != nil {
185
- return nil, resp.Error
186
- }
187
-
188
- return newLimitReadCloser(resp.Output, fetchSizeLimit), nil
189
-}
190
-
191
-// httpFetch attempts to fetch the file at the given URL. Returns
192
-// io.ReadCloser on success, which caller must close.
193
-func httpFetch(ctx context.Context, url string) (io.ReadCloser, error) {
194
- req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
195
- if err != nil {
196
- return nil, fmt.Errorf("http.NewRequest error: %s", err)
197
- }
198
-
199
- req.Header.Set("User-Agent", "go-ipfs")
200
-
201
- resp, err := http.DefaultClient.Do(req)
202
- if err != nil {
203
- return nil, fmt.Errorf("http.DefaultClient.Do error: %s", err)
204
- }
205
-
206
- if resp.StatusCode >= 400 {
207
- defer resp.Body.Close()
208
- mes, err := ioutil.ReadAll(resp.Body)
209
- if err != nil {
210
- return nil, fmt.Errorf("error reading error body: %s", err)
211
- }
212
- return nil, fmt.Errorf("GET %s error: %s: %s", url, resp.Status, string(mes))
213
- }
214
-
215
- return newLimitReadCloser(resp.Body, fetchSizeLimit), nil
216
-}
217
-
218
-func newLimitReadCloser(rc io.ReadCloser, limit int64) io.ReadCloser {
219
- return limitReadCloser{
220
- Reader: io.LimitReader(rc, limit),
221
- Closer: rc,
222
- }
223
-}
224
-
122
// osWithVariant returns the OS name with optional variant.
123
// Currently returns either runtime.GOOS, or "linux-musl".
124
func osWithVariant() (string, error) {
@@ -255,18 +152,21 @@ func osWithVariant() (string, error) {
152
return "linux", nil
153
}
154
258
-// makeArchiveName composes the name of a migration binary archive.
155
+// makeArchivePath composes the path, relative to the distribution site, from which to
156
+// download a binary. The path returned does not contain the distribution site path,
157
+// e.g. "/ipns/dist.ipfs.io/", since that is know to the fetcher.
158
//
260
-// The archive name is in the format: name_version_osv-GOARCH.atype
261
-// Example: ipfs-10-to-11_v1.8.0_darwin-amd64.tar.gz
262
-func makeArchiveName(name, ver, atype string) string {
263
- return fmt.Sprintf("%s_%s_%s-%s.%s", name, ver, runtime.GOOS, runtime.GOARCH, atype)
264
-}
265
-
266
-// makeIpfsPath composes the name ipfs path location to download a migration
267
-// binary from the distribution site.
159
+// Returns the archive path and the base name.
160
+//
161
+// The ipfs path format is: distribution/version/archiveName
162
+// - distribution is the name of a distribution, such as "go-ipfs"
163
+// - version is the version to fetch, such as "v0.8.0-rc2"
164
+// - archiveName is formatted as name_version_osv-GOARCH.atype, such as
165
+// "go-ipfs_v0.8.0-rc2_linux-amd64.tar.gz"
166
//
269
-// The ipfs path format: distBaseCID/rootdir/version/name/archive
270
-func makeIpfsPath(dist, ver, arcName string) string {
271
- return fmt.Sprintf("%s/%s/%s/%s", ipfsDistPath, dist, ver, arcName)
167
+// This would form the path:
168
+// go-ipfs/v0.8.0/go-ipfs_v0.8.0_linux-amd64.tar.gz
169
+func makeArchivePath(dist, name, ver, atype string) (string, string) {
170
+ arcName := fmt.Sprintf("%s_%s_%s-%s.%s", name, ver, runtime.GOOS, runtime.GOARCH, atype)
171
+ return fmt.Sprintf("%s/%s/%s", dist, ver, arcName), arcName
172
}
repo/fsrepo/migrations/fetch_test.go
+97
-77
@@ -3,17 +3,63 @@ package migrations
3
import (
4
"bufio"
5
"context"
6
+ "fmt"
7
+ "io"
8
"io/ioutil"
9
+ "net/http"
10
+ "net/http/httptest"
11
"os"
12
"path"
13
"strings"
14
"testing"
15
)
16
13
-func TestSetIpfsDistPath(t *testing.T) {
17
+func createTestServer() *httptest.Server {
18
+ reqHandler := func(w http.ResponseWriter, r *http.Request) {
19
+ defer r.Body.Close()
20
+ if strings.Contains(r.URL.Path, "not-here") {
21
+ http.NotFound(w, r)
22
+ } else if strings.HasSuffix(r.URL.Path, "versions") {
23
+ fmt.Fprint(w, "v1.0.0\nv1.1.0\nv1.1.2\nv2.0.0-rc1\n2.0.0\nv2.0.1\n")
24
+ } else if strings.HasSuffix(r.URL.Path, ".tar.gz") {
25
+ createFakeArchive(r.URL.Path, false, w)
26
+ } else if strings.HasSuffix(r.URL.Path, "zip") {
27
+ createFakeArchive(r.URL.Path, true, w)
28
+ } else {
29
+ http.NotFound(w, r)
30
+ }
31
+ }
32
+ return httptest.NewServer(http.HandlerFunc(reqHandler))
33
+}
34
+
35
+func createFakeArchive(name string, archZip bool, w io.Writer) {
36
+ fileName := strings.Split(path.Base(name), "_")[0]
37
+ root := path.Base(path.Dir(path.Dir(name)))
38
+
39
+ // Simulate fetching go-ipfs, which has "ipfs" as the name in the archive.
40
+ if fileName == "go-ipfs" {
41
+ fileName = "ipfs"
42
+ }
43
+
44
+ var err error
45
+ if archZip {
46
+ err = writeZip(root, fileName, "FAKE DATA", w)
47
+ } else {
48
+ err = writeTarGzip(root, fileName, "FAKE DATA", w)
49
+ }
50
+ if err != nil {
51
+ panic(err)
52
+ }
53
+}
54
+
55
+func TestSetDistPath(t *testing.T) {
56
+ f1 := NewHttpFetcher()
57
+ f2 := NewHttpFetcher()
58
+ mf := NewMultiFetcher(f1, f2)
59
+
60
os.Unsetenv(envIpfsDistPath)
15
- SetIpfsDistPath("")
16
- if ipfsDistPath != ipfsDist {
61
+ mf.SetDistPath(GetDistPathEnv(""))
62
+ if f1.distPath != IpnsIpfsDist {
63
t.Error("did not set default dist path")
64
}
65
@@ -24,17 +70,30 @@ func TestSetIpfsDistPath(t *testing.T) {
70
}
71
defer func() {
72
os.Unsetenv(envIpfsDistPath)
27
- SetIpfsDistPath("")
73
}()
74
30
- SetIpfsDistPath("")
31
- if ipfsDistPath != testDist {
75
+ mf.SetDistPath(GetDistPathEnv(""))
76
+ if f1.distPath != testDist {
77
+ t.Error("did not set dist path from environ")
78
+ }
79
+ if f2.distPath != testDist {
80
+ t.Error("did not set dist path from environ")
81
+ }
82
+
83
+ mf.SetDistPath(GetDistPathEnv("ignored"))
84
+ if f1.distPath != testDist {
85
+ t.Error("did not set dist path from environ")
86
+ }
87
+ if f2.distPath != testDist {
88
t.Error("did not set dist path from environ")
89
}
90
91
testDist = "/unit/test/dist2"
36
- SetIpfsDistPath(testDist)
37
- if ipfsDistPath != testDist {
92
+ mf.SetDistPath(testDist)
93
+ if f1.distPath != testDist {
94
+ t.Error("did not set dist path")
95
+ }
96
+ if f2.distPath != testDist {
97
t.Error("did not set dist path")
98
}
99
}
@@ -43,8 +102,12 @@ func TestHttpFetch(t *testing.T) {
102
ctx, cancel := context.WithCancel(context.Background())
103
defer cancel()
104
46
- url := gatewayURL + path.Join(ipfsDistPath, distFSRM, distVersions)
47
- rc, err := httpFetch(ctx, url)
105
+ fetcher := NewHttpFetcher()
106
+ ts := createTestServer()
107
+ defer ts.Close()
108
+ fetcher.SetGateway(ts.URL)
109
+
110
+ rc, err := fetcher.Fetch(ctx, "/versions")
111
if err != nil {
112
t.Fatal(err)
113
}
@@ -60,73 +123,18 @@ func TestHttpFetch(t *testing.T) {
123
t.Fatal("could not read versions:", err)
124
}
125
63
- if len(out) < 14 {
126
+ if len(out) < 6 {
127
t.Fatal("do not get all expected data")
128
}
129
if out[0] != "v1.0.0" {
130
t.Fatal("expected v1.0.0 as first line, got", out[0])
131
}
132
70
- // Check bad URL
71
- _, err = httpFetch(ctx, "")
72
- if err == nil {
73
- t.Fatal("expected error")
74
- }
75
-
76
- // Check unreachable URL
77
- _, err = httpFetch(ctx, "http://127.0.0.123:65510")
78
- if err == nil || !strings.HasSuffix(err.Error(), "connection refused") {
79
- t.Fatal("expected 'connection refused' error")
80
- }
81
-
133
// Check not found
83
- url = gatewayURL + path.Join(ipfsDistPath, distFSRM, "no_such_file")
84
- _, err = httpFetch(ctx, url)
134
+ _, err = fetcher.Fetch(ctx, "/no_such_file")
135
if err == nil || !strings.Contains(err.Error(), "404") {
136
t.Fatal("expected error 404")
137
}
88
-
89
-}
90
-
91
-func TestIpfsFetch(t *testing.T) {
92
- _, err := ApiEndpoint("")
93
- if err != nil {
94
- t.Skip("skipped - local ipfs daemon not available")
95
- }
96
-
97
- ctx, cancel := context.WithCancel(context.Background())
98
- defer cancel()
99
-
100
- url := path.Join(ipfsDistPath, distFSRM, distVersions)
101
- rc, err := ipfsFetch(ctx, url)
102
- if err != nil {
103
- t.Fatal(err)
104
- }
105
- defer rc.Close()
106
-
107
- var out []string
108
- scan := bufio.NewScanner(rc)
109
- for scan.Scan() {
110
- out = append(out, scan.Text())
111
- }
112
- err = scan.Err()
113
- if err != nil {
114
- t.Fatal("could not read versions:", err)
115
- }
116
-
117
- if len(out) < 14 {
118
- t.Fatal("do not get all expected data")
119
- }
120
- if out[0] != "v1.0.0" {
121
- t.Fatal("expected v1.0.0 as first line, got", out[0])
122
- }
123
-
124
- // Check bad URL
125
- url = path.Join(ipfsDistPath, distFSRM, "no_such_file")
126
- _, err = ipfsFetch(ctx, url)
127
- if err == nil || !strings.Contains(err.Error(), "no link") {
128
- t.Fatal("expected 'no link' error, got:", err)
129
- }
138
}
139
140
func TestFetchBinary(t *testing.T) {
@@ -139,13 +147,18 @@ func TestFetchBinary(t *testing.T) {
147
ctx, cancel := context.WithCancel(context.Background())
148
defer cancel()
149
142
- vers, err := DistVersions(ctx, distFSRM, false)
150
+ fetcher := NewHttpFetcher()
151
+ ts := createTestServer()
152
+ defer ts.Close()
153
+ fetcher.SetGateway(ts.URL)
154
+
155
+ vers, err := DistVersions(ctx, fetcher, distFSRM, false)
156
if err != nil {
157
t.Fatal(err)
158
}
159
t.Log("latest version of", distFSRM, "is", vers[len(vers)-1])
160
148
- bin, err := FetchBinary(ctx, distFSRM, vers[0], distFSRM, "", tmpDir)
161
+ bin, err := FetchBinary(ctx, fetcher, distFSRM, vers[0], "", tmpDir)
162
if err != nil {
163
t.Fatal(err)
164
}
@@ -157,7 +170,7 @@ func TestFetchBinary(t *testing.T) {
170
171
t.Log("downloaded and unpacked", fi.Size(), "byte file:", fi.Name())
172
160
- bin, err = FetchBinary(ctx, "go-ipfs", "v0.3.5", "", "ipfs", tmpDir)
173
+ bin, err = FetchBinary(ctx, fetcher, "go-ipfs", "v0.3.5", "ipfs", tmpDir)
174
if err != nil {
175
t.Fatal(err)
176
}
@@ -170,11 +183,18 @@ func TestFetchBinary(t *testing.T) {
183
t.Log("downloaded and unpacked", fi.Size(), "byte file:", fi.Name())
184
185
// Check error is destination already exists and is not directory
173
- _, err = FetchBinary(ctx, "go-ipfs", "v0.3.5", "", "ipfs", bin)
186
+ _, err = FetchBinary(ctx, fetcher, "go-ipfs", "v0.3.5", "ipfs", bin)
187
if !os.IsExist(err) {
175
- t.Fatal("expected 'exists' error")
188
+ t.Fatal("expected 'exists' error, got", err)
189
}
190
191
+ _, err = FetchBinary(ctx, fetcher, "go-ipfs", "v0.3.5", "ipfs", tmpDir)
192
+ if !os.IsExist(err) {
193
+ t.Error("expected 'exists' error, got:", err)
194
+ }
195
+
196
+ os.Remove(path.Join(tmpDir, "ipfs"))
197
+
198
// Check error creating temp download directory
199
err = os.Chmod(tmpDir, 0555)
200
if err != nil {
@@ -184,9 +204,9 @@ func TestFetchBinary(t *testing.T) {
204
if err != nil {
205
panic(err)
206
}
187
- _, err = FetchBinary(ctx, "go-ipfs", "v0.3.5", "", "ipfs", tmpDir)
207
+ _, err = FetchBinary(ctx, fetcher, "go-ipfs", "v0.3.5", "ipfs", tmpDir)
208
if !os.IsPermission(err) {
189
- t.Error("expected 'permission'error")
209
+ t.Error("expected 'permission' error, got:", err)
210
}
211
err = os.Setenv("TMPDIR", "/tmp")
212
if err != nil {
@@ -198,13 +218,13 @@ func TestFetchBinary(t *testing.T) {
218
}
219
220
// Check error if failure to fetch due to bad dist
201
- _, err = FetchBinary(ctx, "no-such-dist", "v0.3.5", "", "ipfs", tmpDir)
221
+ _, err = FetchBinary(ctx, fetcher, "not-here", "v0.3.5", "ipfs", tmpDir)
222
if err == nil || !strings.Contains(err.Error(), "Not Found") {
203
- t.Error("expected 'Not Found' error")
223
+ t.Error("expected 'Not Found' error, got:", err)
224
}
225
226
// Check error if failure to unpack archive
207
- _, err = FetchBinary(ctx, "go-ipfs", "v0.3.5", "", "not-such-bin", tmpDir)
227
+ _, err = FetchBinary(ctx, fetcher, "go-ipfs", "v0.3.5", "not-such-bin", tmpDir)
228
if err == nil || err.Error() != "no binary found in archive" {
229
t.Error("expected 'no binary found in archive' error")
230
}
repo/fsrepo/migrations/fetcher.go
new
+96
@@ -0,0 +1,96 @@
1
+package migrations
2
+
3
+import (
4
+ "context"
5
+ "io"
6
+ "os"
7
+ "strings"
8
+)
9
+
10
+const (
11
+ // Current dirstibution to fetch migrations from
12
+ CurrentIpfsDist = "/ipfs/Qme8pJhBidEUXRdpcWLGR2fkG5kdwVnaMh3kabjfP8zz7Y"
13
+ // Distribution IPNS path. Default for fetchers.
14
+ IpnsIpfsDist = "/ipns/dist.ipfs.io"
15
+
16
+ // Distribution environ variable
17
+ envIpfsDistPath = "IPFS_DIST_PATH"
18
+)
19
+
20
+type Fetcher interface {
21
+ // Fetch attempts to fetch the file at the given ipfs path.
22
+ // Returns io.ReadCloser on success, which caller must close.
23
+ Fetch(ctx context.Context, filePath string) (io.ReadCloser, error)
24
+ // SetDistPath sets the path to the distribution site for a Fetcher
25
+ SetDistPath(distPath string)
26
+}
27
+
28
+// MultiFetcher holds multiple Fetchers and provides a Fetch that tries each
29
+// until one succeeds.
30
+type MultiFetcher struct {
31
+ fetchers []Fetcher
32
+}
33
+
34
+type limitReadCloser struct {
35
+ io.Reader
36
+ io.Closer
37
+}
38
+
39
+// NewMultiFetcher creates a MultiFetcher with the given Fetchers. The
40
+// Fetchers are tried in order ther passed to this function.
41
+func NewMultiFetcher(f ...Fetcher) Fetcher {
42
+ mf := &MultiFetcher{
43
+ fetchers: make([]Fetcher, len(f)),
44
+ }
45
+ copy(mf.fetchers, f)
46
+ return mf
47
+}
48
+
49
+// Fetch attempts to fetch the file at each of its fetchers until one succeeds.
50
+// Returns io.ReadCloser on success, which caller must close.
51
+func (f *MultiFetcher) Fetch(ctx context.Context, ipfsPath string) (rc io.ReadCloser, err error) {
52
+ for _, fetcher := range f.fetchers {
53
+ rc, err = fetcher.Fetch(ctx, ipfsPath)
54
+ if err == nil {
55
+ // Transferred using this fetcher
56
+ return
57
+ }
58
+ }
59
+ return
60
+}
61
+
62
+// SetDistPath sets the path to the distribution site for all fetchers
63
+func (f *MultiFetcher) SetDistPath(distPath string) {
64
+ if !strings.HasPrefix(distPath, "/") {
65
+ distPath = "/" + distPath
66
+ }
67
+ for _, fetcher := range f.fetchers {
68
+ fetcher.SetDistPath(distPath)
69
+ }
70
+}
71
+
72
+// NewLimitReadCloser returns a new io.ReadCloser with the reader wrappen in a
73
+// io.LimitedReader limited to reading the amount specified.
74
+func NewLimitReadCloser(rc io.ReadCloser, limit int64) io.ReadCloser {
75
+ return limitReadCloser{
76
+ Reader: io.LimitReader(rc, limit),
77
+ Closer: rc,
78
+ }
79
+}
80
+
81
+// GetDistPathEnv returns the IPFS path to the distribution site, using
82
+// the value of environ variable specified by envIpfsDistPath. If the environ
83
+// variable is not set, then returns the provided distPath, and if that is not set
84
+// then returns the IPNS path.
85
+//
86
+// To get the IPFS path of the latest distribution, if not overriddin by the
87
+// environ variable: GetDistPathEnv(CurrentIpfsDist)
88
+func GetDistPathEnv(distPath string) string {
89
+ if dist := os.Getenv(envIpfsDistPath); dist != "" {
90
+ return dist
91
+ }
92
+ if distPath == "" {
93
+ return IpnsIpfsDist
94
+ }
95
+ return distPath
96
+}
repo/fsrepo/migrations/httpfetcher.go
new
+91
@@ -0,0 +1,91 @@
1
+package migrations
2
+
3
+import (
4
+ "context"
5
+ "fmt"
6
+ "io"
7
+ "io/ioutil"
8
+ "net/http"
9
+ "net/url"
10
+ "path"
11
+ "strings"
12
+)
13
+
14
+const (
15
+ defaultGatewayURL = "https://ipfs.io"
16
+ defaultFetchLimit = 1024 * 1024 * 512
17
+)
18
+
19
+// HttpFetcher fetches files over HTTP
20
+type HttpFetcher struct {
21
+ gateway string
22
+ distPath string
23
+ limit int64
24
+}
25
+
26
+var _ Fetcher = (*HttpFetcher)(nil)
27
+
28
+// NewHttpFetcher creates a new HttpFetcher
29
+func NewHttpFetcher() *HttpFetcher {
30
+ return &HttpFetcher{
31
+ gateway: defaultGatewayURL,
32
+ distPath: IpnsIpfsDist,
33
+ limit: defaultFetchLimit,
34
+ }
35
+}
36
+
37
+// SetGateway sets the gateway URL
38
+func (f *HttpFetcher) SetGateway(gatewayURL string) error {
39
+ gwURL, err := url.Parse(gatewayURL)
40
+ if err != nil {
41
+ return err
42
+ }
43
+ f.gateway = gwURL.String()
44
+ return nil
45
+}
46
+
47
+// SetDistPath sets the path to the distribution site.
48
+func (f *HttpFetcher) SetDistPath(distPath string) {
49
+ if !strings.HasPrefix(distPath, "/") {
50
+ distPath = "/" + distPath
51
+ }
52
+ f.distPath = distPath
53
+}
54
+
55
+// SetFetchLimit sets the download size limit. A value of 0 means no limit.
56
+func (f *HttpFetcher) SetFetchLimit(limit int64) {
57
+ f.limit = limit
58
+}
59
+
60
+// Fetch attempts to fetch the file at the given path, from the distribution
61
+// site configured for this HttpFetcher. Returns io.ReadCloser on success,
62
+// which caller must close.
63
+func (f *HttpFetcher) Fetch(ctx context.Context, filePath string) (io.ReadCloser, error) {
64
+ gwURL := f.gateway + path.Join(f.distPath, filePath)
65
+
66
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, gwURL, nil)
67
+ if err != nil {
68
+ return nil, fmt.Errorf("http.NewRequest error: %s", err)
69
+ }
70
+
71
+ req.Header.Set("User-Agent", "go-ipfs")
72
+
73
+ resp, err := http.DefaultClient.Do(req)
74
+ if err != nil {
75
+ return nil, fmt.Errorf("http.DefaultClient.Do error: %s", err)
76
+ }
77
+
78
+ if resp.StatusCode >= 400 {
79
+ defer resp.Body.Close()
80
+ mes, err := ioutil.ReadAll(resp.Body)
81
+ if err != nil {
82
+ return nil, fmt.Errorf("error reading error body: %s", err)
83
+ }
84
+ return nil, fmt.Errorf("GET %s error: %s: %s", gwURL, resp.Status, string(mes))
85
+ }
86
+
87
+ if f.limit != 0 {
88
+ return NewLimitReadCloser(resp.Body, f.limit), nil
89
+ }
90
+ return resp.Body, nil
91
+}
repo/fsrepo/migrations/ipfsdir.go
-18
@@ -10,7 +10,6 @@ import (
10
"strings"
11
"time"
12
13
- api "github.com/ipfs/go-ipfs-api"
13
"github.com/mitchellh/go-homedir"
14
)
15
@@ -52,23 +51,6 @@ func ApiEndpoint(ipfsDir string) (string, error) {
51
return parts[2] + ":" + parts[4], nil
52
}
53
55
-// ApiShell creates a new ipfs api shell and checks that it is up. If the shell
56
-// is available, then the shell and ipfs version are returned.
57
-func ApiShell(ipfsDir string) (*api.Shell, string, error) {
58
- apiEp, err := ApiEndpoint("")
59
- if err != nil {
60
- return nil, "", err
61
- }
62
- sh := api.NewShell(apiEp)
63
- sh.SetTimeout(shellUpTimeout)
64
- ver, _, err := sh.Version()
65
- if err != nil {
66
- return nil, "", errors.New("ipfs api shell not up")
67
- }
68
- sh.SetTimeout(0)
69
- return sh, ver, nil
70
-}
71
-
54
// IpfsDir returns the path of the ipfs directory. If dir specified, then
55
// returns the expanded version dir. If dir is "", then return the directory
56
// set by IPFS_PATH, or if IPFS_PATH is not set, then return the default
repo/fsrepo/migrations/ipfsdir_test.go
-7
@@ -212,11 +212,4 @@ func TestApiEndpoint(t *testing.T) {
212
if val2 != val {
213
t.Fatal("expected", val, "got", val2)
214
}
215
-
216
- _, _, err = ApiShell(fakeIpfs)
217
- if err != nil {
218
- if err.Error() != "ipfs api shell not up" {
219
- t.Fatal("expected 'ipfs api shell not up' error")
220
- }
221
- }
215
}
repo/fsrepo/migrations/migrations.go
+9
-6
@@ -21,7 +21,7 @@ const (
21
22
// RunMigration finds, downloads, and runs the individual migrations needed to
23
// migrate the repo from its current version to the target version.
24
-func RunMigration(ctx context.Context, targetVer int, ipfsDir string) error {
24
+func RunMigration(ctx context.Context, fetcher Fetcher, targetVer int, ipfsDir string, allowDowngrade bool) error {
25
ipfsDir, err := CheckIpfsDir(ipfsDir)
26
if err != nil {
27
return err
@@ -34,6 +34,9 @@ func RunMigration(ctx context.Context, targetVer int, ipfsDir string) error {
34
// repo already at target version number
35
return nil
36
}
37
+ if fromVer > targetVer && !allowDowngrade {
38
+ return fmt.Errorf("downgrade not allowed from %d to %d", fromVer, targetVer)
39
+ }
40
41
log.Print("Looking for suitable migration binaries.")
42
@@ -59,7 +62,7 @@ func RunMigration(ctx context.Context, targetVer int, ipfsDir string) error {
62
}
63
defer os.RemoveAll(tmpDir)
64
62
- fetched, err := fetchMigrations(ctx, missing, tmpDir)
65
+ fetched, err := fetchMigrations(ctx, fetcher, missing, tmpDir)
66
if err != nil {
67
log.Print("Failed to download migrations.")
68
return err
@@ -156,7 +159,7 @@ func runMigration(ctx context.Context, binPath, ipfsDir string, revert bool) err
159
160
// fetchMigrations downloads the requested migrations, and returns a slice with
161
// the paths of each binary, in the same order specified by needed.
159
-func fetchMigrations(ctx context.Context, needed []string, destDir string) ([]string, error) {
162
+func fetchMigrations(ctx context.Context, fetcher Fetcher, needed []string, destDir string) ([]string, error) {
163
osv, err := osWithVariant()
164
if err != nil {
165
return nil, err
@@ -173,13 +176,13 @@ func fetchMigrations(ctx context.Context, needed []string, destDir string) ([]st
176
log.Printf("Downloading migration: %s...", name)
177
go func(i int, name string) {
178
defer wg.Done()
176
- distDir := path.Join(distMigsRoot, name)
177
- ver, err := LatestDistVersion(ctx, distDir)
179
+ dist := path.Join(distMigsRoot, name)
180
+ ver, err := LatestDistVersion(ctx, fetcher, dist, false)
181
if err != nil {
182
log.Printf("could not get latest version of migration %s: %s", name, err)
183
return
184
}
182
- loc, err := FetchBinary(ctx, distDir, ver, name, name, destDir)
185
+ loc, err := FetchBinary(ctx, fetcher, dist, ver, name, destDir)
186
if err != nil {
187
log.Printf("could not download %s: %s", name, err)
188
return
repo/fsrepo/migrations/migrations_test.go
+52
-12
@@ -3,15 +3,12 @@ package migrations
3
import (
4
"context"
5
"io/ioutil"
6
- "net/http"
6
"os"
7
"path"
8
"strings"
9
"testing"
10
)
11
13
-const testIPFSDistPath = "/ipfs/Qme8pJhBidEUXRdpcWLGR2fkG5kdwVnaMh3kabjfP8zz7Y"
14
-
12
func TestFindMigrations(t *testing.T) {
13
tmpDir, err := ioutil.TempDir("", "migratetest")
14
if err != nil {
@@ -118,14 +115,11 @@ func TestFetchMigrations(t *testing.T) {
115
ctx, cancel := context.WithCancel(context.Background())
116
defer cancel()
117
121
- SetIpfsDistPath(testIPFSDistPath)
122
- _, err := LatestDistVersion(ctx, "ipfs-1-to-2")
123
- if err != nil {
124
- if strings.Contains(err.Error(), http.StatusText(http.StatusNotFound)) {
125
- t.Skip("skip - migrations not yet available on distribution site")
126
- }
127
- t.Fatal(err)
128
- }
118
+ fetcher := NewHttpFetcher()
119
+ fetcher.SetDistPath(CurrentIpfsDist)
120
+ ts := createTestServer()
121
+ defer ts.Close()
122
+ fetcher.SetGateway(ts.URL)
123
124
tmpDir, err := ioutil.TempDir("", "migratetest")
125
if err != nil {
@@ -134,7 +128,7 @@ func TestFetchMigrations(t *testing.T) {
128
defer os.RemoveAll(tmpDir)
129
130
needed := []string{"ipfs-1-to-2", "ipfs-2-to-3"}
137
- fetched, err := fetchMigrations(ctx, needed, tmpDir)
131
+ fetched, err := fetchMigrations(ctx, fetcher, needed, tmpDir)
132
if err != nil {
133
t.Fatal(err)
134
}
@@ -147,6 +141,52 @@ func TestFetchMigrations(t *testing.T) {
141
}
142
}
143
144
+func TestRunMigrations(t *testing.T) {
145
+ var err error
146
+ fakeHome, err = ioutil.TempDir("", "testhome")
147
+ if err != nil {
148
+ panic(err)
149
+ }
150
+ defer os.RemoveAll(fakeHome)
151
+
152
+ os.Setenv("HOME", fakeHome)
153
+ fakeIpfs := path.Join(fakeHome, ".ipfs")
154
+
155
+ err = os.Mkdir(fakeIpfs, os.ModePerm)
156
+ if err != nil {
157
+ panic(err)
158
+ }
159
+
160
+ testVer := 11
161
+ err = WriteRepoVersion(fakeIpfs, testVer)
162
+ if err != nil {
163
+ t.Fatal(err)
164
+ }
165
+
166
+ ctx, cancel := context.WithCancel(context.Background())
167
+ defer cancel()
168
+
169
+ fetcher := NewHttpFetcher()
170
+ fetcher.SetDistPath(CurrentIpfsDist)
171
+ ts := createTestServer()
172
+ defer ts.Close()
173
+ fetcher.SetGateway(ts.URL)
174
+
175
+ targetVer := 9
176
+
177
+ err = RunMigration(ctx, fetcher, targetVer, fakeIpfs, false)
178
+ if err == nil || !strings.HasPrefix(err.Error(), "downgrade not allowed") {
179
+ t.Fatal("expected 'downgrade not alloed' error")
180
+ }
181
+
182
+ err = RunMigration(ctx, fetcher, targetVer, fakeIpfs, true)
183
+ if err != nil {
184
+ if !strings.HasPrefix(err.Error(), "migration ipfs-10-to-11 failed") {
185
+ t.Fatal(err)
186
+ }
187
+ }
188
+}
189
+
190
func createFakeBin(from, to int, tmpDir string) {
191
migPath := path.Join(tmpDir, ExeName(migrationName(from, to)))
192
emptyFile, err := os.Create(migPath)
repo/fsrepo/migrations/unpack.go
+6
-6
@@ -30,13 +30,13 @@ func unpackArchive(arcPath, atype, root, name, out string) error {
30
func unpackTgz(arcPath, root, name, out string) error {
31
fi, err := os.Open(arcPath)
32
if err != nil {
33
- return fmt.Errorf("cannot open archive file: %s", err)
33
+ return fmt.Errorf("cannot open archive file: %w", err)
34
}
35
defer fi.Close()
36
37
gzr, err := gzip.NewReader(fi)
38
if err != nil {
39
- return fmt.Errorf("error opening gzip reader: %s", err)
39
+ return fmt.Errorf("error opening gzip reader: %w", err)
40
}
41
defer gzr.Close()
42
@@ -50,7 +50,7 @@ func unpackTgz(arcPath, root, name, out string) error {
50
if err == io.EOF {
51
break
52
}
53
- return fmt.Errorf("cannot read archive: %s", err)
53
+ return fmt.Errorf("cannot read archive: %w", err)
54
}
55
56
if th.Name == lookFor {
@@ -69,7 +69,7 @@ func unpackTgz(arcPath, root, name, out string) error {
69
func unpackZip(arcPath, root, name, out string) error {
70
zipr, err := zip.OpenReader(arcPath)
71
if err != nil {
72
- return fmt.Errorf("error opening zip reader: %s", err)
72
+ return fmt.Errorf("error opening zip reader: %w", err)
73
}
74
defer zipr.Close()
75
@@ -79,7 +79,7 @@ func unpackZip(arcPath, root, name, out string) error {
79
if fis.Name == lookFor {
80
rc, err := fis.Open()
81
if err != nil {
82
- return fmt.Errorf("error extracting binary from archive: %s", err)
82
+ return fmt.Errorf("error extracting binary from archive: %w", err)
83
}
84
85
bin = rc
@@ -97,7 +97,7 @@ func unpackZip(arcPath, root, name, out string) error {
97
func writeToPath(rc io.Reader, out string) error {
98
binfi, err := os.Create(out)
99
if err != nil {
100
- return fmt.Errorf("error opening tmp bin path '%s': %s", out, err)
100
+ return fmt.Errorf("error creating output file '%s': %w", out, err)
101
}
102
defer binfi.Close()
103
repo/fsrepo/migrations/unpack_test.go
+42
-24
@@ -5,6 +5,7 @@ import (
5
"archive/zip"
6
"bufio"
7
"compress/gzip"
8
+ "io"
9
"io/ioutil"
10
"os"
11
"path"
@@ -49,7 +50,7 @@ func TestUnpackTgz(t *testing.T) {
50
51
testTarGzip := path.Join(tmpDir, "test.tar.gz")
52
testData := "some data"
52
- err = writeTarGzip(testTarGzip, "testroot", "testfile", testData)
53
+ err = writeTarGzipFile(testTarGzip, "testroot", "testfile", testData)
54
if err != nil {
55
panic(err)
56
}
@@ -97,7 +98,7 @@ func TestUnpackZip(t *testing.T) {
98
99
testZip := path.Join(tmpDir, "test.zip")
100
testData := "some data"
100
- err = writeZip(testZip, "testroot", "testfile", testData)
101
+ err = writeZipFile(testZip, "testroot", "testfile", testData)
102
if err != nil {
103
panic(err)
104
}
@@ -125,21 +126,38 @@ func TestUnpackZip(t *testing.T) {
126
}
127
}
128
128
-func writeTarGzip(archName, root, fileName, data string) error {
129
+func writeTarGzipFile(archName, root, fileName, data string) error {
130
archFile, err := os.Create(archName)
131
if err != nil {
132
return err
133
}
134
defer archFile.Close()
134
- wr := bufio.NewWriter(archFile)
135
+ w := bufio.NewWriter(archFile)
136
137
+ err = writeTarGzip(root, fileName, data, w)
138
+ if err != nil {
139
+ return err
140
+ }
141
+ // Flush buffered data to file
142
+ if err = w.Flush(); err != nil {
143
+ return err
144
+ }
145
+ // Close tar file
146
+ if err = archFile.Close(); err != nil {
147
+ return err
148
+ }
149
+ return nil
150
+}
151
+
152
+func writeTarGzip(root, fileName, data string, w io.Writer) error {
153
// gzip writer writes to buffer
137
- gzw := gzip.NewWriter(wr)
154
+ gzw := gzip.NewWriter(w)
155
defer gzw.Close()
156
// tar writer writes to gzip
157
tw := tar.NewWriter(gzw)
158
defer tw.Close()
159
160
+ var err error
161
if fileName != "" {
162
hdr := &tar.Header{
163
Name: path.Join(root, fileName),
@@ -163,26 +181,34 @@ func writeTarGzip(archName, root, fileName, data string) error {
181
if err = gzw.Close(); err != nil {
182
return err
183
}
166
- // Flush buffered data to file
167
- if err = wr.Flush(); err != nil {
168
- return err
169
- }
170
- // Close tar file
171
- if err = archFile.Close(); err != nil {
172
- return err
173
- }
184
return nil
185
}
186
177
-func writeZip(archName, root, fileName, data string) error {
187
+func writeZipFile(archName, root, fileName, data string) error {
188
archFile, err := os.Create(archName)
189
if err != nil {
190
return err
191
}
192
defer archFile.Close()
183
- wr := bufio.NewWriter(archFile)
193
+ w := bufio.NewWriter(archFile)
194
185
- zw := zip.NewWriter(wr)
195
+ err = writeZip(root, fileName, data, w)
196
+ if err != nil {
197
+ return err
198
+ }
199
+ // Flush buffered data to file
200
+ if err = w.Flush(); err != nil {
201
+ return err
202
+ }
203
+ // Close zip file
204
+ if err = archFile.Close(); err != nil {
205
+ return err
206
+ }
207
+ return nil
208
+}
209
+
210
+func writeZip(root, fileName, data string, w io.Writer) error {
211
+ zw := zip.NewWriter(w)
212
defer zw.Close()
213
214
// Write file name
@@ -200,13 +226,5 @@ func writeZip(archName, root, fileName, data string) error {
226
if err = zw.Close(); err != nil {
227
return err
228
}
203
- // Flush buffered data to file
204
- if err = wr.Flush(); err != nil {
205
- return err
206
- }
207
- // Close zip file
208
- if err = archFile.Close(); err != nil {
209
- return err
210
- }
229
return nil
230
}
repo/fsrepo/migrations/versions.go
+10
-6
@@ -18,17 +18,21 @@ const distVersions = "versions"
18
19
// LatestDistVersion returns the latest version, of the specified distribution,
20
// that is available on the distribution site.
21
-func LatestDistVersion(ctx context.Context, dist string) (string, error) {
22
- vs, err := DistVersions(ctx, dist, false)
21
+func LatestDistVersion(ctx context.Context, fetcher Fetcher, dist string, stableOnly bool) (string, error) {
22
+ vs, err := DistVersions(ctx, fetcher, dist, false)
23
if err != nil {
24
return "", err
25
}
26
27
for i := len(vs) - 1; i >= 0; i-- {
28
ver := vs[i]
29
- if !strings.Contains(ver, "-dev") {
30
- return ver, nil
29
+ if stableOnly && strings.Contains(ver, "-rc") {
30
+ continue
31
+ }
32
+ if strings.Contains(ver, "-dev") {
33
+ continue
34
}
35
+ return ver, nil
36
}
37
return "", errors.New("could not find a non dev version")
38
}
@@ -36,8 +40,8 @@ func LatestDistVersion(ctx context.Context, dist string) (string, error) {
40
// DistVersions returns all versions of the specified distribution, that are
41
// available on the distriburion site. List is in ascending order, unless
42
// sortDesc is true.
39
-func DistVersions(ctx context.Context, dist string, sortDesc bool) ([]string, error) {
40
- rc, err := fetch(ctx, path.Join(ipfsDistPath, dist, distVersions))
43
+func DistVersions(ctx context.Context, fetcher Fetcher, dist string, sortDesc bool) ([]string, error) {
44
+ rc, err := fetcher.Fetch(ctx, path.Join(dist, distVersions))
45
if err != nil {
46
return nil, err
47
}
repo/fsrepo/migrations/versions_test.go
+12
-2
@@ -14,7 +14,12 @@ func TestDistVersions(t *testing.T) {
14
ctx, cancel := context.WithCancel(context.Background())
15
defer cancel()
16
17
- vers, err := DistVersions(ctx, testDist, true)
17
+ fetcher := NewHttpFetcher()
18
+ ts := createTestServer()
19
+ defer ts.Close()
20
+ fetcher.SetGateway(ts.URL)
21
+
22
+ vers, err := DistVersions(ctx, fetcher, testDist, true)
23
if err != nil {
24
t.Fatal(err)
25
}
@@ -29,7 +34,12 @@ func TestLatestDistVersion(t *testing.T) {
34
ctx, cancel := context.WithCancel(context.Background())
35
defer cancel()
36
32
- latest, err := LatestDistVersion(ctx, testDist)
37
+ fetcher := NewHttpFetcher()
38
+ //ts := createTestServer()
39
+ //defer ts.Close()
40
+ //fetcher.SetGateway(ts.URL)
41
+
42
+ latest, err := LatestDistVersion(ctx, fetcher, testDist, false)
43
if err != nil {
44
t.Fatal(err)
45
}