@cryptotaxi247 / kubo / commits / c54cdaa1f

Feat/migration ipfs download (#8064)

* Enable downloading migrations over IPFS There are now options in the config file that control how migrations are downloaded. This includes enabling downloading migrations using IPFS by (when migrations are required) spinning up a temporary node for fetching the migrations before running them. There is also a config option to decide what to do with the migrations binaries once they are downloaded (e.g. cache or pin them in your node, or just throw out the data). Co-authored-by: Steven Allen <steven@stebalien.com>

Andrew Gillis committed May 12, 2021 at 09:33 UTC c54cdaa1f87507ba1048c8d450dc43e6db6eb1d7
16 files changed +1186 -27
cmd/ipfs/daemon.go
+56 -2
@@ -4,6 +4,7 @@ import (
4 "errors"
5 _ "expvar"
6 "fmt"
7 + "io/ioutil"
8 "net"
9 "net/http"
10 _ "net/http/pprof"
@@ -268,6 +269,9 @@ func daemonFunc(req *cmds.Request, re cmds.ResponseEmitter, env cmds.Environment
269 }
270 }
271
272 + var cacheMigrations, pinMigrations bool
273 + var fetcher migrations.Fetcher
274 +
275 // acquire the repo lock _before_ constructing a node. we need to make
276 // sure we are permitted to access the resources (datastore, etc.)
277 repo, err := fsrepo.Open(cctx.ConfigRoot)
@@ -288,8 +292,38 @@ func daemonFunc(req *cmds.Request, re cmds.ResponseEmitter, env cmds.Environment
292 return fmt.Errorf("fs-repo requires migration")
293 }
294
291 - // Fetch migrations from current distribution, or location from environ
292 - fetcher := migrations.NewHttpFetcher(migrations.GetDistPathEnv(migrations.CurrentIpfsDist), "", "go-ipfs", 0)
295 + migrationCfg, err := readMigrationConfig(cctx.ConfigRoot)
296 + if err != nil {
297 + return err
298 + }
299 +
300 + fetcher, err = getMigrationFetcher(migrationCfg, &cctx.ConfigRoot)
301 + if err != nil {
302 + return err
303 + }
304 + defer fetcher.Close()
305 +
306 + if migrationCfg.Keep == "cache" {
307 + cacheMigrations = true
308 + } else if migrationCfg.Keep == "pin" {
309 + pinMigrations = true
310 + }
311 +
312 + if cacheMigrations || pinMigrations {
313 + // Create temp directory to store downloaded migration archives
314 + migrations.DownloadDirectory, err = ioutil.TempDir("", "migrations")
315 + if err != nil {
316 + return err
317 + }
318 + // Defer cleanup of download directory so that it gets cleaned up
319 + // if daemon returns early due to error
320 + defer func() {
321 + if migrations.DownloadDirectory != "" {
322 + os.RemoveAll(migrations.DownloadDirectory)
323 + }
324 + }()
325 + }
326 +
327 err = migrations.RunMigration(cctx.Context(), fetcher, fsrepo.RepoVersion, "", false)
328 if err != nil {
329 fmt.Println("The migrations of fs-repo failed:")
@@ -420,6 +454,26 @@ func daemonFunc(req *cmds.Request, re cmds.ResponseEmitter, env cmds.Environment
454 return err
455 }
456
457 + // Add any files downloaded by migration.
458 + if cacheMigrations || pinMigrations {
459 + err = addMigrations(cctx.Context(), node, fetcher, pinMigrations)
460 + if err != nil {
461 + fmt.Fprintln(os.Stderr, "Could not add migragion to IPFS:", err)
462 + }
463 + // Remove download directory so that it does not remain for lifetime of
464 + // daemon or get left behind if daemon has a hard exit
465 + os.RemoveAll(migrations.DownloadDirectory)
466 + migrations.DownloadDirectory = ""
467 + }
468 + if fetcher != nil {
469 + // If there is an error closing the IpfsFetcher, then print error, but
470 + // do not fail because of it.
471 + err = fetcher.Close()
472 + if err != nil {
473 + log.Errorf("error closing IPFS fetcher: %s", err)
474 + }
475 + }
476 +
477 // construct http gateway
478 gwErrc, err := serveHTTPGateway(req, cctx)
479 if err != nil {
cmd/ipfs/migration.go new
+300
@@ -0,0 +1,300 @@
1 +package main
2 +
3 +import (
4 + "context"
5 + "encoding/json"
6 + "errors"
7 + "fmt"
8 + "io"
9 + "io/ioutil"
10 + "net/url"
11 + "os"
12 + "path/filepath"
13 + "strings"
14 +
15 + config "github.com/ipfs/go-ipfs-config"
16 + "github.com/ipfs/go-ipfs-files"
17 + "github.com/ipfs/go-ipfs/core"
18 + "github.com/ipfs/go-ipfs/core/coreapi"
19 + "github.com/ipfs/go-ipfs/repo/fsrepo/migrations"
20 + "github.com/ipfs/go-ipfs/repo/fsrepo/migrations/ipfsfetcher"
21 + coreiface "github.com/ipfs/interface-go-ipfs-core"
22 + "github.com/ipfs/interface-go-ipfs-core/options"
23 + ipath "github.com/ipfs/interface-go-ipfs-core/path"
24 + "github.com/libp2p/go-libp2p-core/peer"
25 +)
26 +
27 +// readMigrationConfig reads the migration config out of the config, avoiding
28 +// reading anything other than the migration section. That way, we're free to
29 +// make arbitrary changes to all _other_ sections in migrations.
30 +func readMigrationConfig(repoRoot string) (*config.Migration, error) {
31 + var cfg struct {
32 + Migration config.Migration
33 + }
34 +
35 + cfgPath, err := config.Filename(repoRoot)
36 + if err != nil {
37 + return nil, err
38 + }
39 +
40 + cfgFile, err := os.Open(cfgPath)
41 + if err != nil {
42 + return nil, err
43 + }
44 + defer cfgFile.Close()
45 +
46 + err = json.NewDecoder(cfgFile).Decode(&cfg)
47 + if err != nil {
48 + return nil, err
49 + }
50 +
51 + switch cfg.Migration.Keep {
52 + case "":
53 + cfg.Migration.Keep = config.DefaultMigrationKeep
54 + case "discard", "cache", "keep":
55 + default:
56 + return nil, errors.New("unknown config value, Migrations.Keep must be 'cache', 'pin', or 'discard'")
57 + }
58 +
59 + if len(cfg.Migration.DownloadSources) == 0 {
60 + cfg.Migration.DownloadSources = config.DefaultMigrationDownloadSources
61 + }
62 +
63 + return &cfg.Migration, nil
64 +}
65 +
66 +func readIpfsConfig(repoRoot *string) (bootstrap []string, peers []peer.AddrInfo) {
67 + if repoRoot == nil {
68 + return
69 + }
70 +
71 + cfgPath, err := config.Filename(*repoRoot)
72 + if err != nil {
73 + fmt.Fprintln(os.Stderr, err)
74 + return
75 + }
76 +
77 + cfgFile, err := os.Open(cfgPath)
78 + if err != nil {
79 + fmt.Fprintln(os.Stderr, err)
80 + return
81 + }
82 + defer cfgFile.Close()
83 +
84 + // Attempt to read bootstrap addresses
85 + var bootstrapCfg struct {
86 + Bootstrap []string
87 + }
88 + err = json.NewDecoder(cfgFile).Decode(&bootstrapCfg)
89 + if err != nil {
90 + fmt.Fprintln(os.Stderr, "cannot read bootstrap peers from config")
91 + } else {
92 + bootstrap = bootstrapCfg.Bootstrap
93 + }
94 +
95 + if _, err = cfgFile.Seek(0, 0); err != nil {
96 + fmt.Fprintln(os.Stderr, err)
97 + }
98 +
99 + // Attempt to read peers
100 + var peeringCfg struct {
101 + Peering config.Peering
102 + }
103 + err = json.NewDecoder(cfgFile).Decode(&peeringCfg)
104 + if err != nil {
105 + fmt.Fprintln(os.Stderr, "cannot read peering from config")
106 + } else {
107 + peers = peeringCfg.Peering.Peers
108 + }
109 +
110 + return
111 +}
112 +
113 +// getMigrationFetcher creates one or more fetchers according to
114 +// config.Migration.DownloadSources. If an IpfsFetcher is required, then
115 +// bootstrap and peer information in read from the config file in repoRoot,
116 +// unless repoRoot is nil.
117 +func getMigrationFetcher(cfg *config.Migration, repoRoot *string) (migrations.Fetcher, error) {
118 + const httpUserAgent = "go-ipfs"
119 +
120 + // Fetch migrations from current distribution, or location from environ
121 + fetchDistPath := migrations.GetDistPathEnv(migrations.CurrentIpfsDist)
122 +
123 + var fetchers []migrations.Fetcher
124 + for _, src := range cfg.DownloadSources {
125 + src := strings.TrimSpace(src)
126 + switch src {
127 + case "IPFS", "ipfs":
128 + bootstrap, peers := readIpfsConfig(repoRoot)
129 + fetchers = append(fetchers, ipfsfetcher.NewIpfsFetcher(fetchDistPath, 0, bootstrap, peers))
130 + case "HTTPS", "https", "HTTP", "http":
131 + fetchers = append(fetchers, migrations.NewHttpFetcher(fetchDistPath, "", httpUserAgent, 0))
132 + default:
133 + u, err := url.Parse(src)
134 + if err != nil {
135 + return nil, fmt.Errorf("bad gateway address: %s", err)
136 + }
137 + switch u.Scheme {
138 + case "":
139 + u.Scheme = "https"
140 + case "https", "http":
141 + default:
142 + return nil, errors.New("bad gateway address: url scheme must be http or https")
143 + }
144 + fetchers = append(fetchers, migrations.NewHttpFetcher(fetchDistPath, u.String(), httpUserAgent, 0))
145 + case "":
146 + // Ignore empty string
147 + }
148 + }
149 + if len(fetchers) == 0 {
150 + return nil, errors.New("no sources specified")
151 + }
152 +
153 + if len(fetchers) == 1 {
154 + return fetchers[0], nil
155 + }
156 +
157 + // Wrap fetchers in a MultiFetcher to try them in order
158 + return migrations.NewMultiFetcher(fetchers...), nil
159 +}
160 +
161 +func addMigrations(ctx context.Context, node *core.IpfsNode, fetcher migrations.Fetcher, pin bool) error {
162 + var fetchers []migrations.Fetcher
163 + if mf, ok := fetcher.(*migrations.MultiFetcher); ok {
164 + fetchers = mf.Fetchers()
165 + } else {
166 + fetchers = []migrations.Fetcher{fetcher}
167 + }
168 +
169 + for _, fetcher := range fetchers {
170 + switch f := fetcher.(type) {
171 + case *ipfsfetcher.IpfsFetcher:
172 + // Add migrations by connecting to temp node and getting from IPFS
173 + err := addMigrationPaths(ctx, node, f.AddrInfo(), f.FetchedPaths(), pin)
174 + if err != nil {
175 + return err
176 + }
177 + case *migrations.HttpFetcher:
178 + // Add the downloaded migration files directly
179 + if migrations.DownloadDirectory != "" {
180 + var paths []string
181 + err := filepath.Walk(migrations.DownloadDirectory, func(filePath string, info os.FileInfo, err error) error {
182 + if info.IsDir() {
183 + return nil
184 + }
185 + paths = append(paths, filePath)
186 + return nil
187 + })
188 + if err != nil {
189 + return err
190 + }
191 + err = addMigrationFiles(ctx, node, paths, pin)
192 + if err != nil {
193 + return err
194 + }
195 + }
196 + default:
197 + return errors.New("Cannot get migrations from unknown fetcher type")
198 + }
199 + }
200 +
201 + return nil
202 +}
203 +
204 +// addMigrationFiles adds the files at paths to IPFS, optionally pinning them
205 +func addMigrationFiles(ctx context.Context, node *core.IpfsNode, paths []string, pin bool) error {
206 + if len(paths) == 0 {
207 + return nil
208 + }
209 + ifaceCore, err := coreapi.NewCoreAPI(node)
210 + if err != nil {
211 + return err
212 + }
213 + ufs := ifaceCore.Unixfs()
214 +
215 + // Add migration files
216 + for _, filePath := range paths {
217 + f, err := os.Open(filePath)
218 + if err != nil {
219 + return err
220 + }
221 +
222 + fi, err := f.Stat()
223 + if err != nil {
224 + return err
225 + }
226 +
227 + ipfsPath, err := ufs.Add(ctx, files.NewReaderStatFile(f, fi), options.Unixfs.Pin(pin))
228 + if err != nil {
229 + return err
230 + }
231 + fmt.Printf("Added migration file %q: %s\n", filepath.Base(filePath), ipfsPath)
232 + }
233 +
234 + return nil
235 +}
236 +
237 +// addMigrationPaths adds the files at paths to IPFS, optionally pinning
238 +// them. This is done after connecting to the peer.
239 +func addMigrationPaths(ctx context.Context, node *core.IpfsNode, peerInfo peer.AddrInfo, paths []ipath.Path, pin bool) error {
240 + if len(paths) == 0 {
241 + return errors.New("nothing downloaded by ipfs fetcher")
242 + }
243 + if len(peerInfo.Addrs) == 0 {
244 + return errors.New("no local swarm address for migration node")
245 + }
246 +
247 + ipfs, err := coreapi.NewCoreAPI(node)
248 + if err != nil {
249 + return err
250 + }
251 +
252 + // Connect to temp node
253 + if err := ipfs.Swarm().Connect(ctx, peerInfo); err != nil {
254 + return fmt.Errorf("could not connect to migration peer %q: %s", peerInfo.ID, err)
255 + }
256 + fmt.Printf("connected to migration peer %q\n", peerInfo)
257 +
258 + if pin {
259 + pinApi := ipfs.Pin()
260 + for _, ipfsPath := range paths {
261 + err := pinApi.Add(ctx, ipfsPath)
262 + if err != nil {
263 + return err
264 + }
265 + fmt.Printf("Added and pinned migration file: %q\n", ipfsPath)
266 + }
267 + return nil
268 + }
269 +
270 + ufs := ipfs.Unixfs()
271 +
272 + // Add migration files
273 + for _, ipfsPath := range paths {
274 + err = ipfsGet(ctx, ufs, ipfsPath)
275 + if err != nil {
276 + return err
277 + }
278 + }
279 +
280 + return nil
281 +}
282 +
283 +func ipfsGet(ctx context.Context, ufs coreiface.UnixfsAPI, ipfsPath ipath.Path) error {
284 + nd, err := ufs.Get(ctx, ipfsPath)
285 + if err != nil {
286 + return err
287 + }
288 + defer nd.Close()
289 +
290 + fnd, ok := nd.(files.File)
291 + if !ok {
292 + return fmt.Errorf("not a file node: %q", ipfsPath)
293 + }
294 + _, err = io.Copy(ioutil.Discard, fnd)
295 + if err != nil {
296 + return fmt.Errorf("cannot read migration: %w", err)
297 + }
298 + fmt.Printf("Added migration file: %q\n", ipfsPath)
299 + return nil
300 +}
cmd/ipfs/migration_test.go new
+312
@@ -0,0 +1,312 @@
1 +package main
2 +
3 +import (
4 + "io/ioutil"
5 + "os"
6 + "path/filepath"
7 + "strings"
8 + "testing"
9 +
10 + config "github.com/ipfs/go-ipfs-config"
11 + "github.com/ipfs/go-ipfs/repo/fsrepo/migrations"
12 + "github.com/ipfs/go-ipfs/repo/fsrepo/migrations/ipfsfetcher"
13 +)
14 +
15 +var testConfig = `
16 +{
17 + "Bootstrap": [
18 + "/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
19 + "/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ"
20 + ],
21 + "Migration": {
22 + "DownloadSources": ["IPFS", "HTTP", "127.0.0.1", "https://127.0.1.1"],
23 + "Keep": "cache"
24 + },
25 + "Peering": {
26 + "Peers": [
27 + {
28 + "ID": "12D3KooWGC6TvWhfapngX6wvJHMYvKpDMXPb3ZnCZ6dMoaMtimQ5",
29 + "Addrs": ["/ip4/127.0.0.1/tcp/4001", "/ip4/127.0.0.1/udp/4001/quic"]
30 + }
31 + ]
32 + }
33 +}
34 +`
35 +
36 +func TestReadMigrationConfigDefaults(t *testing.T) {
37 + tmpDir := makeConfig("{}")
38 + defer os.RemoveAll(tmpDir)
39 +
40 + cfg, err := readMigrationConfig(tmpDir)
41 + if err != nil {
42 + t.Fatal(err)
43 + }
44 +
45 + if cfg.Keep != config.DefaultMigrationKeep {
46 + t.Error("expected default value for Keep")
47 + }
48 +
49 + if len(cfg.DownloadSources) != len(config.DefaultMigrationDownloadSources) {
50 + t.Fatal("expected default number of download sources")
51 + }
52 + for i, src := range config.DefaultMigrationDownloadSources {
53 + if cfg.DownloadSources[i] != src {
54 + t.Errorf("wrong DownloadSource: %s", cfg.DownloadSources[i])
55 + }
56 + }
57 +}
58 +
59 +func TestReadMigrationConfigErrors(t *testing.T) {
60 + tmpDir := makeConfig(`{"Migration": {"Keep": "badvalue"}}`)
61 + defer os.RemoveAll(tmpDir)
62 +
63 + _, err := readMigrationConfig(tmpDir)
64 + if err == nil {
65 + t.Fatal("expected error")
66 + }
67 + if !strings.HasPrefix(err.Error(), "unknown") {
68 + t.Fatal("did not get expected error:", err)
69 + }
70 +
71 + os.RemoveAll(tmpDir)
72 + _, err = readMigrationConfig(tmpDir)
73 + if err == nil {
74 + t.Fatal("expected error")
75 + }
76 +
77 + bootstrap, peers := readIpfsConfig(&tmpDir)
78 + if bootstrap != nil {
79 + t.Error("expected nil bootstrap")
80 + }
81 + if peers != nil {
82 + t.Error("expected nil peers")
83 + }
84 +
85 + tmpDir = makeConfig(`}{`)
86 + defer os.RemoveAll(tmpDir)
87 + _, err = readMigrationConfig(tmpDir)
88 + if err == nil {
89 + t.Fatal("expected error")
90 + }
91 +}
92 +
93 +func TestReadMigrationConfig(t *testing.T) {
94 + tmpDir := makeConfig(testConfig)
95 + defer os.RemoveAll(tmpDir)
96 +
97 + cfg, err := readMigrationConfig(tmpDir)
98 + if err != nil {
99 + t.Fatal(err)
100 + }
101 +
102 + if len(cfg.DownloadSources) != 4 {
103 + t.Fatal("wrong number of DownloadSources")
104 + }
105 + expect := []string{"IPFS", "HTTP", "127.0.0.1", "https://127.0.1.1"}
106 + for i := range expect {
107 + if cfg.DownloadSources[i] != expect[i] {
108 + t.Errorf("wrong DownloadSource at %d", i)
109 + }
110 + }
111 +
112 + if cfg.Keep != "cache" {
113 + t.Error("wrong value for Keep")
114 + }
115 +}
116 +
117 +func TestReadIpfsConfig(t *testing.T) {
118 + tmpDir := makeConfig(testConfig)
119 + defer os.RemoveAll(tmpDir)
120 +
121 + bootstrap, peers := readIpfsConfig(nil)
122 + if bootstrap != nil || peers != nil {
123 + t.Fatal("expected nil ipfs config items")
124 + }
125 +
126 + bootstrap, peers = readIpfsConfig(&tmpDir)
127 + if len(bootstrap) != 2 {
128 + t.Fatal("wrong number of bootstrap addresses")
129 + }
130 + if bootstrap[0] != "/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt" {
131 + t.Fatal("wrong bootstrap address")
132 + }
133 +
134 + if len(peers) != 1 {
135 + t.Fatal("wrong number of peers")
136 + }
137 +
138 + peer := peers[0]
139 + if peer.ID.String() != "12D3KooWGC6TvWhfapngX6wvJHMYvKpDMXPb3ZnCZ6dMoaMtimQ5" {
140 + t.Errorf("wrong ID for first peer")
141 + }
142 + if len(peer.Addrs) != 2 {
143 + t.Error("wrong number of addrs for first peer")
144 + }
145 +}
146 +
147 +func TestReadPartialIpfsConfig(t *testing.T) {
148 + const (
149 + configBadBootstrap = `
150 +{
151 + "Bootstrap": "unreadable",
152 + "Migration": {
153 + "DownloadSources": ["IPFS", "HTTP", "127.0.0.1"],
154 + "Keep": "cache"
155 + },
156 + "Peering": {
157 + "Peers": [
158 + {
159 + "ID": "12D3KooWGC6TvWhfapngX6wvJHMYvKpDMXPb3ZnCZ6dMoaMtimQ5",
160 + "Addrs": ["/ip4/127.0.0.1/tcp/4001", "/ip4/127.0.0.1/udp/4001/quic"]
161 + }
162 + ]
163 + }
164 +}
165 +`
166 + configBadPeers = `
167 +{
168 + "Bootstrap": [
169 + "/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
170 + "/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ"
171 + ],
172 + "Migration": {
173 + "DownloadSources": ["IPFS", "HTTP", "127.0.0.1"],
174 + "Keep": "cache"
175 + },
176 + "Peering": "Unreadable-data"
177 +}
178 +`
179 + )
180 +
181 + tmpDir := makeConfig(configBadBootstrap)
182 + defer os.RemoveAll(tmpDir)
183 +
184 + bootstrap, peers := readIpfsConfig(&tmpDir)
185 + if bootstrap != nil {
186 + t.Fatal("expected nil bootstrap")
187 + }
188 + if len(peers) != 1 {
189 + t.Fatal("wrong number of peers")
190 + }
191 + if len(peers[0].Addrs) != 2 {
192 + t.Error("wrong number of addrs for first peer")
193 + }
194 + os.RemoveAll(tmpDir)
195 +
196 + tmpDir = makeConfig(configBadPeers)
197 + defer os.RemoveAll(tmpDir)
198 +
199 + bootstrap, peers = readIpfsConfig(&tmpDir)
200 + if peers != nil {
201 + t.Fatal("expected nil peers")
202 + }
203 + if len(bootstrap) != 2 {
204 + t.Fatal("wrong number of bootstrap addresses")
205 + }
206 + if bootstrap[0] != "/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt" {
207 + t.Fatal("wrong bootstrap address")
208 + }
209 +}
210 +
211 +func makeConfig(configData string) string {
212 + tmpDir, err := ioutil.TempDir("", "migration_test")
213 + if err != nil {
214 + panic(err)
215 + }
216 +
217 + cfgFile, err := os.Create(filepath.Join(tmpDir, "config"))
218 + if err != nil {
219 + panic(err)
220 + }
221 + if _, err = cfgFile.Write([]byte(configData)); err != nil {
222 + panic(err)
223 + }
224 + if err = cfgFile.Close(); err != nil {
225 + panic(err)
226 + }
227 + return tmpDir
228 +}
229 +
230 +func TestGetMigrationFetcher(t *testing.T) {
231 + var f migrations.Fetcher
232 + var err error
233 +
234 + cfg := &config.Migration{}
235 +
236 + cfg.DownloadSources = []string{"ftp://bad.gateway.io"}
237 + _, err = getMigrationFetcher(cfg, nil)
238 + if err == nil || !strings.HasPrefix(err.Error(), "bad gateway addr") {
239 + t.Fatal("Expected bad gateway address error, got:", err)
240 + }
241 +
242 + cfg.DownloadSources = []string{"::bad.gateway.io"}
243 + _, err = getMigrationFetcher(cfg, nil)
244 + if err == nil || !strings.HasPrefix(err.Error(), "bad gateway addr") {
245 + t.Fatal("Expected bad gateway address error, got:", err)
246 + }
247 +
248 + cfg.DownloadSources = []string{"http://localhost"}
249 + f, err = getMigrationFetcher(cfg, nil)
250 + if err != nil {
251 + t.Fatal(err)
252 + }
253 + if _, ok := f.(*migrations.HttpFetcher); !ok {
254 + t.Fatal("expected HttpFetcher")
255 + }
256 +
257 + cfg.DownloadSources = []string{"ipfs"}
258 + f, err = getMigrationFetcher(cfg, nil)
259 + if err != nil {
260 + t.Fatal(err)
261 + }
262 + if _, ok := f.(*ipfsfetcher.IpfsFetcher); !ok {
263 + t.Fatal("expected IpfsFetcher")
264 + }
265 +
266 + cfg.DownloadSources = []string{"http"}
267 + f, err = getMigrationFetcher(cfg, nil)
268 + if err != nil {
269 + t.Fatal(err)
270 + }
271 + if _, ok := f.(*migrations.HttpFetcher); !ok {
272 + t.Fatal("expected HttpFetcher")
273 + }
274 +
275 + cfg.DownloadSources = []string{"IPFS", "HTTPS"}
276 + f, err = getMigrationFetcher(cfg, nil)
277 + if err != nil {
278 + t.Fatal(err)
279 + }
280 + mf, ok := f.(*migrations.MultiFetcher)
281 + if !ok {
282 + t.Fatal("expected MultiFetcher")
283 + }
284 + if mf.Len() != 2 {
285 + t.Fatal("expected 2 fetchers in MultiFetcher")
286 + }
287 +
288 + cfg.DownloadSources = []string{"ipfs", "https", "some.domain.io"}
289 + f, err = getMigrationFetcher(cfg, nil)
290 + if err != nil {
291 + t.Fatal(err)
292 + }
293 + mf, ok = f.(*migrations.MultiFetcher)
294 + if !ok {
295 + t.Fatal("expected MultiFetcher")
296 + }
297 + if mf.Len() != 3 {
298 + t.Fatal("expected 3 fetchers in MultiFetcher")
299 + }
300 +
301 + cfg.DownloadSources = nil
302 + _, err = getMigrationFetcher(cfg, nil)
303 + if err == nil {
304 + t.Fatal("expected error when no sources specified")
305 + }
306 +
307 + cfg.DownloadSources = []string{"", ""}
308 + _, err = getMigrationFetcher(cfg, nil)
309 + if err == nil {
310 + t.Fatal("expected error when empty string fetchers specified")
311 + }
312 +}
docs/config.md
+19
@@ -172,6 +172,9 @@ does (e.g, `"1d2h4m40.01s"`).
172 - [`Ipns.RepublishPeriod`](#ipnsrepublishperiod)
173 - [`Ipns.RecordLifetime`](#ipnsrecordlifetime)
174 - [`Ipns.ResolveCacheSize`](#ipnsresolvecachesize)
175 +- [`Migration`](#migration)
176 + - [`Migration.DownloadSources`](#migrationdownloadsources)
177 + - [`Migration.Keep`](#migrationkeep)
178 - [`Mounts`](#mounts)
179 - [`Mounts.IPFS`](#mountsipfs)
180 - [`Mounts.IPNS`](#mountsipns)
@@ -809,6 +812,22 @@ Default: `128`
812
813 Type: `integer` (non-negative, 0 means the default)
814
815 +## `Migration`
816 +
817 +Migration configures how migrations are downloaded and if the downloads are added to IPFS locally.
818 +
819 +### `Migration.DownloadSources`
820 +
821 +Sources in order of preference, where "IPFS" means use IPFS and "HTTPS" means use default gateways. Any other values are interpreted as hostnames for custom gateways. An empty list means "use default sources".
822 +
823 +Default: `["HTTPS", "IPFS"]`
824 +
825 +### `Migration.Keep`
826 +
827 +Specifies whether or not to keep the migration after downloading it. Options are "discard", "cache", "pin". Empty string for default.
828 +
829 +Default: `cache`
830 +
831 ## `Mounts`
832
833 FUSE mount point configuration options.
go.mod
+2 -2
@@ -102,10 +102,10 @@ require (
102 go.opencensus.io v0.23.0
103 go.uber.org/fx v1.13.1
104 go.uber.org/zap v1.16.0
105 - golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2
105 + golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b
106 golang.org/x/lint v0.0.0-20201208152925-83fdc39ff7b5 // indirect
107 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
108 - golang.org/x/sys v0.0.0-20210426080607-c94f62235c83
108 + golang.org/x/sys v0.0.0-20210503173754-0981d6026fa6
109 )
110
111 go 1.14
go.sum
+4 -2
@@ -1288,8 +1288,9 @@ golang.org/x/crypto v0.0.0-20200510223506-06a226fb4e37/go.mod h1:LzIPMQfyMNhhGPh
1288 golang.org/x/crypto v0.0.0-20200602180216-279210d13fed/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
1289 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
1290 golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
1291 -golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2 h1:It14KIkyBFYkHkwZ7k45minvA9aorojkyjGk9KJ5B/w=
1291 golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
1292 +golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b h1:7mWr3k41Qtv8XlltBkDkl8LoP3mpSgBW8BUoxtEdbXg=
1293 +golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
1294 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
1295 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
1296 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -1441,8 +1442,9 @@ golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7w
1442 golang.org/x/sys v0.0.0-20210303074136-134d130e1a04/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
1443 golang.org/x/sys v0.0.0-20210309074719-68d13333faf2/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
1444 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
1444 -golang.org/x/sys v0.0.0-20210426080607-c94f62235c83 h1:kHSDPqCtsHZOg0nVylfTo20DDhE9gG4Y0jn7hKQ0QAM=
1445 golang.org/x/sys v0.0.0-20210426080607-c94f62235c83/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
1446 +golang.org/x/sys v0.0.0-20210503173754-0981d6026fa6 h1:cdsMqa2nXzqlgs183pHxtvoVwU7CyzaCTAUOg94af4c=
1447 +golang.org/x/sys v0.0.0-20210503173754-0981d6026fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
1448 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E=
1449 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
1450 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
repo/fsrepo/fsrepo.go
+1 -1
@@ -14,7 +14,6 @@ import (
14 keystore "github.com/ipfs/go-ipfs-keystore"
15 repo "github.com/ipfs/go-ipfs/repo"
16 "github.com/ipfs/go-ipfs/repo/common"
17 - "github.com/ipfs/go-ipfs/repo/fsrepo/migrations"
17 dir "github.com/ipfs/go-ipfs/thirdparty/dir"
18
19 ds "github.com/ipfs/go-datastore"
@@ -23,6 +22,7 @@ import (
22 config "github.com/ipfs/go-ipfs-config"
23 serialize "github.com/ipfs/go-ipfs-config/serialize"
24 util "github.com/ipfs/go-ipfs-util"
25 + "github.com/ipfs/go-ipfs/repo/fsrepo/migrations"
26 logging "github.com/ipfs/go-log"
27 homedir "github.com/mitchellh/go-homedir"
28 ma "github.com/multiformats/go-multiaddr"
repo/fsrepo/migrations/fetch.go
+26 -5
@@ -14,6 +14,12 @@ import (
14 "strings"
15 )
16
17 +// DownloadDirectory can be set as the location for FetchBinary to save the
18 +// downloaded archive file in. If not set, then FetchBinary saves the archive
19 +// in a temporary directory that is removed after the contents of the archive
20 +// is extracted.
21 +var DownloadDirectory string
22 +
23 // FetchBinary downloads an archive from the distribution site and unpacks it.
24 //
25 // The base name of the binary inside the archive may differ from the base
@@ -68,12 +74,27 @@ func FetchBinary(ctx context.Context, fetcher Fetcher, dist, ver, binName, out s
74 }
75 }
76
71 - // Create temp directory to store download
72 - tmpDir, err := ioutil.TempDir("", arcName)
73 - if err != nil {
74 - return "", err
77 + tmpDir := DownloadDirectory
78 + if tmpDir != "" {
79 + fi, err = os.Stat(tmpDir)
80 + if err != nil {
81 + return "", err
82 + }
83 + if !fi.IsDir() {
84 + return "", &os.PathError{
85 + Op: "FetchBinary",
86 + Path: tmpDir,
87 + Err: os.ErrExist,
88 + }
89 + }
90 + } else {
91 + // Create temp directory to store download
92 + tmpDir, err = ioutil.TempDir("", arcName)
93 + if err != nil {
94 + return "", err
95 + }
96 + defer os.RemoveAll(tmpDir)
97 }
76 - defer os.RemoveAll(tmpDir)
98
99 atype := "tar.gz"
100 if runtime.GOOS == "windows" {
repo/fsrepo/migrations/fetch_test.go
+5 -5
@@ -102,21 +102,21 @@ func TestHttpFetch(t *testing.T) {
102 }
103 defer rc.Close()
104
105 - var out []string
105 + var lines []string
106 scan := bufio.NewScanner(rc)
107 for scan.Scan() {
108 - out = append(out, scan.Text())
108 + lines = append(lines, scan.Text())
109 }
110 err = scan.Err()
111 if err != nil {
112 t.Fatal("could not read versions:", err)
113 }
114
115 - if len(out) < 6 {
115 + if len(lines) < 6 {
116 t.Fatal("do not get all expected data")
117 }
118 - if out[0] != "v1.0.0" {
119 - t.Fatal("expected v1.0.0 as first line, got", out[0])
118 + if lines[0] != "v1.0.0" {
119 + t.Fatal("expected v1.0.0 as first line, got", lines[0])
120 }
121
122 // Check not found
repo/fsrepo/migrations/fetcher.go
+28 -5
@@ -4,6 +4,8 @@ import (
4 "context"
5 "io"
6 "os"
7 +
8 + "github.com/hashicorp/go-multierror"
9 )
10
11 const (
@@ -20,6 +22,8 @@ type Fetcher interface {
22 // Fetch attempts to fetch the file at the given ipfs path.
23 // Returns io.ReadCloser on success, which caller must close.
24 Fetch(ctx context.Context, filePath string) (io.ReadCloser, error)
25 + // Close performs any cleanup after the fetcher is not longer needed.
26 + Close() error
27 }
28
29 // MultiFetcher holds multiple Fetchers and provides a Fetch that tries each
@@ -45,15 +49,34 @@ func NewMultiFetcher(f ...Fetcher) Fetcher {
49
50 // Fetch attempts to fetch the file at each of its fetchers until one succeeds.
51 // Returns io.ReadCloser on success, which caller must close.
48 -func (f *MultiFetcher) Fetch(ctx context.Context, ipfsPath string) (rc io.ReadCloser, err error) {
52 +func (f *MultiFetcher) Fetch(ctx context.Context, ipfsPath string) (io.ReadCloser, error) {
53 + var errs error
54 for _, fetcher := range f.fetchers {
50 - rc, err = fetcher.Fetch(ctx, ipfsPath)
55 + rc, err := fetcher.Fetch(ctx, ipfsPath)
56 if err == nil {
52 - // Transferred using this fetcher
53 - return
57 + return rc, nil
58 + }
59 + errs = multierror.Append(errs, err)
60 + }
61 + return nil, errs
62 +}
63 +
64 +func (f *MultiFetcher) Close() error {
65 + var errs error
66 + for _, fetcher := range f.fetchers {
67 + if err := fetcher.Close(); err != nil {
68 + errs = multierror.Append(errs, err)
69 }
70 }
56 - return
71 + return errs
72 +}
73 +
74 +func (f *MultiFetcher) Len() int {
75 + return len(f.fetchers)
76 +}
77 +
78 +func (f *MultiFetcher) Fetchers() []Fetcher {
79 + return f.fetchers
80 }
81
82 // NewLimitReadCloser returns a new io.ReadCloser with the reader wrappen in a
repo/fsrepo/migrations/httpfetcher.go
+7 -1
@@ -12,6 +12,7 @@ import (
12
13 const (
14 defaultGatewayURL = "https://ipfs.io"
15 + // Default maximum download size
16 defaultFetchLimit = 1024 * 1024 * 512
17 )
18
@@ -49,7 +50,7 @@ func NewHttpFetcher(distPath, gateway, userAgent string, fetchLimit int64) *Http
50 }
51
52 if fetchLimit != 0 {
52 - if fetchLimit == -1 {
53 + if fetchLimit < 0 {
54 fetchLimit = 0
55 }
56 f.limit = fetchLimit
@@ -63,6 +64,7 @@ func NewHttpFetcher(distPath, gateway, userAgent string, fetchLimit int64) *Http
64 // which caller must close.
65 func (f *HttpFetcher) Fetch(ctx context.Context, filePath string) (io.ReadCloser, error) {
66 gwURL := f.gateway + path.Join(f.distPath, filePath)
67 + fmt.Printf("Fetching with HTTP: %q\n", gwURL)
68
69 req, err := http.NewRequestWithContext(ctx, http.MethodGet, gwURL, nil)
70 if err != nil {
@@ -92,3 +94,7 @@ func (f *HttpFetcher) Fetch(ctx context.Context, filePath string) (io.ReadCloser
94 }
95 return resp.Body, nil
96 }
97 +
98 +func (f *HttpFetcher) Close() error {
99 + return nil
100 +}
repo/fsrepo/migrations/ipfsfetcher/ipfsfetcher.go new
+279
@@ -0,0 +1,279 @@
1 +package ipfsfetcher
2 +
3 +import (
4 + "context"
5 + "fmt"
6 + "io"
7 + "io/ioutil"
8 + "net/url"
9 + "os"
10 + "path"
11 + "strings"
12 + "sync"
13 +
14 + "github.com/ipfs/go-ipfs-config"
15 + files "github.com/ipfs/go-ipfs-files"
16 + "github.com/ipfs/go-ipfs/core"
17 + "github.com/ipfs/go-ipfs/core/coreapi"
18 + "github.com/ipfs/go-ipfs/core/node/libp2p"
19 + "github.com/ipfs/go-ipfs/repo/fsrepo"
20 + "github.com/ipfs/go-ipfs/repo/fsrepo/migrations"
21 + iface "github.com/ipfs/interface-go-ipfs-core"
22 + "github.com/ipfs/interface-go-ipfs-core/options"
23 + ipath "github.com/ipfs/interface-go-ipfs-core/path"
24 + peer "github.com/libp2p/go-libp2p-core/peer"
25 +)
26 +
27 +const (
28 + // Default maximum download size
29 + defaultFetchLimit = 1024 * 1024 * 512
30 +
31 + tempNodeTcpAddr = "/ip4/127.0.0.1/tcp/0"
32 +)
33 +
34 +type IpfsFetcher struct {
35 + distPath string
36 + limit int64
37 + bootstrap []string
38 + peers []peer.AddrInfo
39 +
40 + openOnce sync.Once
41 + openErr error
42 + closeOnce sync.Once
43 + closeErr error
44 +
45 + ipfs iface.CoreAPI
46 + ipfsTmpDir string
47 + ipfsStopFunc func()
48 +
49 + fetched []ipath.Path
50 + mutex sync.Mutex
51 +
52 + addrInfo peer.AddrInfo
53 +}
54 +
55 +// NewIpfsFetcher creates a new IpfsFetcher
56 +//
57 +// Specifying "" for distPath sets the default IPNS path.
58 +// Specifying 0 for fetchLimit sets the default, -1 means no limit.
59 +func NewIpfsFetcher(distPath string, fetchLimit int64, bootstrap []string, peers []peer.AddrInfo) *IpfsFetcher {
60 + f := &IpfsFetcher{
61 + limit: defaultFetchLimit,
62 + distPath: migrations.LatestIpfsDist,
63 + bootstrap: bootstrap,
64 + peers: peers,
65 + }
66 +
67 + if distPath != "" {
68 + if !strings.HasPrefix(distPath, "/") {
69 + distPath = "/" + distPath
70 + }
71 + f.distPath = distPath
72 + }
73 +
74 + if fetchLimit != 0 {
75 + if fetchLimit < 0 {
76 + fetchLimit = 0
77 + }
78 + f.limit = fetchLimit
79 + }
80 +
81 + return f
82 +}
83 +
84 +// Fetch attempts to fetch the file at the given path, from the distribution
85 +// site configured for this HttpFetcher. Returns io.ReadCloser on success,
86 +// which caller must close.
87 +func (f *IpfsFetcher) Fetch(ctx context.Context, filePath string) (io.ReadCloser, error) {
88 + // Initialize and start IPFS node on first call to Fetch, since the fetcher
89 + // may be created by not used.
90 + f.openOnce.Do(func() {
91 + f.ipfsTmpDir, f.openErr = initTempNode(ctx, f.bootstrap, f.peers)
92 + if f.openErr != nil {
93 + return
94 + }
95 +
96 + f.openErr = f.startTempNode(ctx)
97 + })
98 +
99 + fmt.Printf("Fetching with IPFS: %q\n", filePath)
100 +
101 + if f.openErr != nil {
102 + return nil, f.openErr
103 + }
104 +
105 + iPath, err := parsePath(path.Join(f.distPath, filePath))
106 + if err != nil {
107 + return nil, err
108 + }
109 +
110 + nd, err := f.ipfs.Unixfs().Get(ctx, iPath)
111 + if err != nil {
112 + return nil, err
113 + }
114 +
115 + f.recordFetched(iPath)
116 +
117 + fileNode, ok := nd.(files.File)
118 + if !ok {
119 + return nil, fmt.Errorf("%q is not a file", filePath)
120 + }
121 +
122 + if f.limit != 0 {
123 + return migrations.NewLimitReadCloser(fileNode, f.limit), nil
124 + }
125 + return fileNode, nil
126 +}
127 +
128 +func (f *IpfsFetcher) Close() error {
129 + f.closeOnce.Do(func() {
130 + if f.ipfsStopFunc != nil {
131 + // Tell ipfs node to stop and wait for it to stop
132 + f.ipfsStopFunc()
133 + }
134 +
135 + if f.ipfsTmpDir != "" {
136 + // Remove the temp ipfs dir
137 + f.closeErr = os.RemoveAll(f.ipfsTmpDir)
138 + }
139 + })
140 + return f.closeErr
141 +}
142 +
143 +func (f *IpfsFetcher) AddrInfo() peer.AddrInfo {
144 + return f.addrInfo
145 +}
146 +
147 +// FetchedPaths returns the IPFS paths of all items fetched by this fetcher
148 +func (f *IpfsFetcher) FetchedPaths() []ipath.Path {
149 + f.mutex.Lock()
150 + defer f.mutex.Unlock()
151 + return f.fetched
152 +}
153 +
154 +func (f *IpfsFetcher) recordFetched(fetchedPath ipath.Path) {
155 + // Mutex protects against update by concurrent calls to Fetch
156 + f.mutex.Lock()
157 + defer f.mutex.Unlock()
158 + f.fetched = append(f.fetched, fetchedPath)
159 +}
160 +
161 +func initTempNode(ctx context.Context, bootstrap []string, peers []peer.AddrInfo) (string, error) {
162 + identity, err := config.CreateIdentity(ioutil.Discard, []options.KeyGenerateOption{
163 + options.Key.Type(options.Ed25519Key),
164 + })
165 + if err != nil {
166 + return "", err
167 + }
168 + cfg, err := config.InitWithIdentity(identity)
169 + if err != nil {
170 + return "", err
171 + }
172 +
173 + // create temporary ipfs directory
174 + dir, err := ioutil.TempDir("", "ipfs-temp")
175 + if err != nil {
176 + return "", fmt.Errorf("failed to get temp dir: %s", err)
177 + }
178 +
179 + // configure the temporary node
180 + cfg.Routing.Type = "dhtclient"
181 +
182 + // Disable listening for inbound connections
183 + cfg.Addresses.Gateway = []string{}
184 + cfg.Addresses.API = []string{}
185 + cfg.Addresses.Swarm = []string{tempNodeTcpAddr}
186 +
187 + if len(bootstrap) != 0 {
188 + cfg.Bootstrap = bootstrap
189 + }
190 +
191 + if len(peers) != 0 {
192 + cfg.Peering.Peers = peers
193 + }
194 +
195 + // Assumes that repo plugins are already loaded
196 + err = fsrepo.Init(dir, cfg)
197 + if err != nil {
198 + os.RemoveAll(dir)
199 + return "", fmt.Errorf("failed to initialize ephemeral node: %s", err)
200 + }
201 +
202 + return dir, nil
203 +}
204 +
205 +func (f *IpfsFetcher) startTempNode(ctx context.Context) error {
206 + // Open the repo
207 + r, err := fsrepo.Open(f.ipfsTmpDir)
208 + if err != nil {
209 + return err
210 + }
211 +
212 + // Create a new lifetime context that is used to stop the temp ipfs node
213 + ctxIpfsLife, cancel := context.WithCancel(context.Background())
214 +
215 + // Construct the node
216 + node, err := core.NewNode(ctxIpfsLife, &core.BuildCfg{
217 + Online: true,
218 + Routing: libp2p.DHTClientOption,
219 + Repo: r,
220 + })
221 + if err != nil {
222 + cancel()
223 + r.Close()
224 + return err
225 + }
226 +
227 + ipfs, err := coreapi.NewCoreAPI(node)
228 + if err != nil {
229 + cancel()
230 + return err
231 + }
232 +
233 + stopFunc := func() {
234 + // Tell ipfs to stop
235 + cancel()
236 + // Wait until ipfs is stopped
237 + <-node.Context().Done()
238 +
239 + fmt.Println("migration peer", node.Identity, "shutdown")
240 + }
241 +
242 + addrs, err := ipfs.Swarm().LocalAddrs(ctx)
243 + if err != nil {
244 + // Failure to get the local swarm address only means that the
245 + // downloaded migrations cannot be fetched through the temporary node.
246 + // So, print the error message and keep going.
247 + fmt.Fprintln(os.Stderr, "cannot get local swarm address:", err)
248 + }
249 +
250 + f.addrInfo = peer.AddrInfo{
251 + ID: node.Identity,
252 + Addrs: addrs,
253 + }
254 +
255 + f.ipfs = ipfs
256 + f.ipfsStopFunc = stopFunc
257 +
258 + return nil
259 +}
260 +
261 +func parsePath(fetchPath string) (ipath.Path, error) {
262 + ipfsPath := ipath.New(fetchPath)
263 + if ipfsPath.IsValid() == nil {
264 + return ipfsPath, nil
265 + }
266 +
267 + u, err := url.Parse(fetchPath)
268 + if err != nil {
269 + return nil, fmt.Errorf("%q could not be parsed: %s", fetchPath, err)
270 + }
271 +
272 + switch proto := u.Scheme; proto {
273 + case "ipfs", "ipld", "ipns":
274 + ipfsPath = ipath.New(path.Join("/", proto, u.Host, u.Path))
275 + default:
276 + return nil, fmt.Errorf("%q is not an IPFS path", fetchPath)
277 + }
278 + return ipfsPath, ipfsPath.IsValid()
279 +}
repo/fsrepo/migrations/ipfsfetcher/ipfsfetcher_test.go new
+144
@@ -0,0 +1,144 @@
1 +package ipfsfetcher
2 +
3 +import (
4 + "bufio"
5 + "context"
6 + "fmt"
7 + "os"
8 + "path/filepath"
9 + "testing"
10 +
11 + "github.com/ipfs/go-ipfs/plugin/loader"
12 + "github.com/ipfs/go-ipfs/repo/fsrepo/migrations"
13 +)
14 +
15 +func init() {
16 + err := setupPlugins()
17 + if err != nil {
18 + panic(err)
19 + }
20 +}
21 +
22 +func TestIpfsFetcher(t *testing.T) {
23 + skipUnlessEpic(t)
24 +
25 + ctx, cancel := context.WithCancel(context.Background())
26 + defer cancel()
27 +
28 + fetcher := NewIpfsFetcher("", 0, nil, nil)
29 + defer fetcher.Close()
30 +
31 + rc, err := fetcher.Fetch(ctx, "go-ipfs/versions")
32 + if err != nil {
33 + t.Fatal(err)
34 + }
35 + defer rc.Close()
36 +
37 + var lines []string
38 + scan := bufio.NewScanner(rc)
39 + for scan.Scan() {
40 + lines = append(lines, scan.Text())
41 + }
42 + err = scan.Err()
43 + if err != nil {
44 + t.Fatal("could not read versions:", err)
45 + }
46 +
47 + if len(lines) < 6 {
48 + t.Fatal("do not get all expected data")
49 + }
50 + if lines[0] != "v0.3.2" {
51 + t.Fatal("expected v1.0.0 as first line, got", lines[0])
52 + }
53 +
54 + // Check not found
55 + _, err = fetcher.Fetch(ctx, "/no_such_file")
56 + if err == nil {
57 + t.Fatal("expected error 404")
58 + }
59 +
60 +}
61 +
62 +func TestInitIpfsFetcher(t *testing.T) {
63 + ctx, cancel := context.WithCancel(context.Background())
64 + defer cancel()
65 +
66 + f := NewIpfsFetcher("", 0, nil, nil)
67 + defer f.Close()
68 +
69 + // Init ipfs repo
70 + f.ipfsTmpDir, f.openErr = initTempNode(ctx, f.bootstrap, f.peers)
71 + if f.openErr != nil {
72 + t.Fatalf("failed to initialize ipfs node: %s", f.openErr)
73 + }
74 +
75 + // Start ipfs node
76 + f.openErr = f.startTempNode(ctx)
77 + if f.openErr != nil {
78 + t.Errorf("failed to start ipfs node: %s", f.openErr)
79 + return
80 + }
81 +
82 + var stopFuncCalled bool
83 + stopFunc := f.ipfsStopFunc
84 + f.ipfsStopFunc = func() {
85 + stopFuncCalled = true
86 + stopFunc()
87 + }
88 +
89 + addrInfo := f.AddrInfo()
90 + if string(addrInfo.ID) == "" {
91 + t.Error("AddInfo ID not set")
92 + }
93 + if len(addrInfo.Addrs) == 0 {
94 + t.Error("AddInfo Addrs not set")
95 + }
96 + t.Log("Temp node listening on:", addrInfo.Addrs)
97 +
98 + err := f.Close()
99 + if err != nil {
100 + t.Fatalf("failed to close fetcher: %s", err)
101 + }
102 +
103 + if stopFunc != nil && !stopFuncCalled {
104 + t.Error("Close did not call stop function")
105 + }
106 +
107 + err = f.Close()
108 + if err != nil {
109 + t.Fatalf("failed to close fetcher 2nd time: %s", err)
110 + }
111 +}
112 +
113 +func skipUnlessEpic(t *testing.T) {
114 + if os.Getenv("IPFS_EPIC_TEST") == "" {
115 + t.SkipNow()
116 + }
117 +}
118 +
119 +func setupPlugins() error {
120 + defaultPath, err := migrations.IpfsDir("")
121 + if err != nil {
122 + return err
123 + }
124 +
125 + // Load plugins. This will skip the repo if not available.
126 + plugins, err := loader.NewPluginLoader(filepath.Join(defaultPath, "plugins"))
127 + if err != nil {
128 + return fmt.Errorf("error loading plugins: %w", err)
129 + }
130 +
131 + if err := plugins.Initialize(); err != nil {
132 + // Need to ignore errors here because plugins may already be loaded when
133 + // run from ipfs daemon.
134 + return fmt.Errorf("error initializing plugins: %w", err)
135 + }
136 +
137 + if err := plugins.Inject(); err != nil {
138 + // Need to ignore errors here because plugins may already be loaded when
139 + // run from ipfs daemon.
140 + return fmt.Errorf("error injecting plugins: %w", err)
141 + }
142 +
143 + return nil
144 +}
repo/fsrepo/migrations/migrations.go
+2 -1
@@ -26,7 +26,7 @@ func RunMigration(ctx context.Context, fetcher Fetcher, targetVer int, ipfsDir s
26 if err != nil {
27 return err
28 }
29 - fromVer, err := repoVersion(ipfsDir)
29 + fromVer, err := RepoVersion(ipfsDir)
30 if err != nil {
31 return fmt.Errorf("could not get repo version: %s", err)
32 }
@@ -69,6 +69,7 @@ func RunMigration(ctx context.Context, fetcher Fetcher, targetVer int, ipfsDir s
69 logger.Print("Failed to download migrations.")
70 return err
71 }
72 +
73 for i := range missing {
74 binPaths[missing[i]] = fetched[i]
75 }
repo/fsrepo/migrations/migrations_test.go
+1 -2
@@ -157,8 +157,7 @@ func TestFetchMigrations(t *testing.T) {
157 }
158
159 func TestRunMigrations(t *testing.T) {
160 - var err error
161 - fakeHome, err = ioutil.TempDir("", "testhome")
160 + fakeHome, err := ioutil.TempDir("", "testhome")
161 if err != nil {
162 panic(err)
163 }
repo/fsrepo/migrations/unpack.go
-1
@@ -23,7 +23,6 @@ func unpackArchive(arcPath, atype, root, name, out string) error {
23 if err != nil {
24 return err
25 }
26 - os.Remove(arcPath)
26 return nil
27 }
28