6863543637

Committed 14 Nov 2023 12:18PM UTC coverage: 61.774% (+0.1%) from 61.668%

Build # 6863543637

Build Type

Pull #1137

github

Committed by

altergui

Commit Message

ipfs: refactor removing very old clutter and abstractions

* data.Storage.Init now needs no args
* move DataMockTest into its own package datamock
* assert at compile time that datamock.DataMockTest and ipfs.Handler satisfy data.Storage
* move fsrepo.IsInitialized check into startNode
* remove ancient abstraction types.DataStore
* fixup "ipfs: add EnableLocalDiscovery option": move field EnableLocalDiscovery into ipfs.Handler

Pull Request Pull Request #1137: feat/ipfs refactor

Run Details

42 of 45 new or added lines in 7 files covered. (93.33%)

4 existing lines in 2 files now uncovered.

14625 of 23675 relevant lines covered (61.77%)

31034.21 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.32

/data/ipfs/init.go

package ipfs

import (
        "context"
        "errors"
        "fmt"
        "io"
        "os"
        "path"
        "sync"

        coreiface "github.com/ipfs/boxo/coreiface"
        ihelper "github.com/ipfs/boxo/ipld/unixfs/importer/helpers"
        ipfscid "github.com/ipfs/go-cid"
        "github.com/ipfs/kubo/commands"
        "github.com/ipfs/kubo/config"
        ipfscore "github.com/ipfs/kubo/core"
        ipfsapi "github.com/ipfs/kubo/core/coreapi"
        "github.com/ipfs/kubo/plugin/loader"
        "github.com/ipfs/kubo/repo"
        "github.com/ipfs/kubo/repo/fsrepo"
        "github.com/ipfs/kubo/repo/fsrepo/migrations"
        "github.com/ipfs/kubo/repo/fsrepo/migrations/ipfsfetcher"
        "github.com/multiformats/go-multicodec"
        "github.com/multiformats/go-multihash"

        "go.vocdoni.io/dvote/log"
)

var ConfigRoot string

// ChunkerTypeSize is the chunker type used by IPFS to calculate to build the DAG.
const ChunkerTypeSize = "size-262144"

func init() {
        // Initialize the DAG builder with offline exchange and the correct CID format
        format := ipfscid.V1Builder{
                Codec:  uint64(multicodec.DagJson),
                MhType: uint64(multihash.SHA2_256),
        }
        dAGbuilder = ihelper.DagBuilderParams{
                Dagserv:    dAG(),
                RawLeaves:  false,
                Maxlinks:   ihelper.DefaultLinksPerBlock,
                NoCopy:     false,
                CidBuilder: &format,
        }
}

// Init initializes the IPFS node and repository.
func initRepository(enableLocalDiscovery bool) error {
        daemonLocked, err := fsrepo.LockedByOtherProcess(ConfigRoot)
        if err != nil {
                return err
        }
        log.Info("checking if daemon is running")
        if daemonLocked {
                log.Debug("ipfs daemon is running")
                return fmt.Errorf("ipfs daemon is running. please stop it to run this command")
        }

        if err := os.MkdirAll(ConfigRoot, 0750); err != nil {
                return err
        }

        if err := installDatabasePlugins(); err != nil {
                return err
        }
        _, err = doInit(io.Discard, ConfigRoot, 2048, enableLocalDiscovery)
        return err
}

// StartNode starts the IPFS node.
func (i *Handler) startNode() (*ipfscore.IpfsNode, coreiface.CoreAPI, error) {
        log.Infow("starting IPFS node", "config", ConfigRoot)

        // check if needs init
        if !fsrepo.IsInitialized(ConfigRoot) {
                if err := initRepository(i.EnableLocalDiscovery); err != nil {
                        return nil, nil, err
                }
        }

        r, err := fsrepo.Open(ConfigRoot)
        if errors.Is(err, fsrepo.ErrNeedMigration) {
                log.Warn("Found outdated ipfs repo, migrations need to be run.")
                r, err = runMigrationsAndOpen(ConfigRoot)
        }
        if err != nil {
                return nil, nil, err
        }
        cfg := &ipfscore.BuildCfg{
                Repo:      r,
                Online:    true,
                Permanent: true,
        }

        // We use node.Cancel to stop it instead.
        ctx := context.Background()

        node, err := ipfscore.NewNode(ctx, cfg)
        if err != nil {
                return nil, nil, err
        }
        node.IsDaemon = true
        node.IsOnline = true

        api, err := ipfsapi.NewCoreAPI(node)
        if err != nil {
                return nil, nil, err
        }
        return node, api, nil
}

// runMigrationsAndOpen fetches and applies migrations just like upstream kubo does
// and returns fsrepo.Open(ConfigRoot)
func runMigrationsAndOpen(ConfigRoot string) (repo.Repo, error) {
        // Read Migration section of IPFS config
        migrationCfg, err := migrations.ReadMigrationConfig(ConfigRoot, "")
        if err != nil {
                return nil, err
        }

        // Define function to create IPFS fetcher.  Do not supply an
        // already-constructed IPFS fetcher, because this may be expensive and
        // not needed according to migration config. Instead, supply a function
        // to construct the particular IPFS fetcher implementation used here,
        // which is called only if an IPFS fetcher is needed.
        newIpfsFetcher := func(distPath string) migrations.Fetcher {
                return ipfsfetcher.NewIpfsFetcher(distPath, 0, &ConfigRoot, "")
        }

        // Fetch migrations from current distribution, or location from environ
        fetchDistPath := migrations.GetDistPathEnv(migrations.CurrentIpfsDist)

        // Create fetchers according to migrationCfg.DownloadSources
        fetcher, err := migrations.GetMigrationFetcher(migrationCfg.DownloadSources,
                fetchDistPath, newIpfsFetcher)
        if err != nil {
                return nil, err
        }
        defer fetcher.Close()

        if migrationCfg.Keep == "cache" || migrationCfg.Keep == "pin" {
                // Create temp directory to store downloaded migration archives
                migrations.DownloadDirectory, err = os.MkdirTemp("", "migrations")
                if err != nil {
                        return nil, err
                }
                // Defer cleanup of download directory so that it gets cleaned up
                // if daemon returns early due to error
                defer func() {
                        if migrations.DownloadDirectory != "" {
                                _ = os.RemoveAll(migrations.DownloadDirectory)
                        }
                }()
        }

        err = migrations.RunMigration(context.TODO(), fetcher, fsrepo.RepoVersion, ConfigRoot, false)
        if err != nil {
                return nil, fmt.Errorf("migrations of ipfs-repo failed: %w", err)
        }

        return fsrepo.Open(ConfigRoot)
}

// CmdCtx returns a commands.Context for the given node and repo path.
func cmdCtx(node *ipfscore.IpfsNode, repoPath string) commands.Context {
        return commands.Context{
                ConfigRoot: repoPath,
                ConstructNode: func() (*ipfscore.IpfsNode, error) {
                        return node, nil
                },
        }
}

var installDatabasePlugins = sync.OnceValue(func() error {
        loader, err := loader.NewPluginLoader("")
        if err != nil {
                return err
        }
        if err := loader.Initialize(); err != nil {
                return err
        }
        if err := loader.Inject(); err != nil {
                return err
        }
        return nil
})

func doInit(out io.Writer, repoRoot string, nBitsForKeypair int, enableLocalDiscovery bool) (*config.Config, error) {
        log.Infow("initializing new IPFS repository", "root", repoRoot)
        if err := checkWritable(repoRoot); err != nil {
                return nil, err
        }

        if fsrepo.IsInitialized(repoRoot) {
                return nil, fmt.Errorf("repo exists")
        }

        conf, err := config.Init(out, nBitsForKeypair)
        if err != nil {
                return nil, err
        }

        // Apply `server` configuration profile:
        // Disables local host discovery, recommended when running IPFS on machines with public IPv4 addresses
        // Prevent from scanning local networks which can trigger netscan alerts.
        // See: https://github.com/ipfs/kubo/issues/7985
        conf.Discovery.MDNS.Enabled = false
        conf.Swarm.DisableNatPortMap = true
        conf.Swarm.AddrFilters = []string{
                "/ip4/10.0.0.0/ipcidr/8",
                "/ip4/100.64.0.0/ipcidr/10",
                "/ip4/169.254.0.0/ipcidr/16",
                "/ip4/172.16.0.0/ipcidr/12",
                "/ip4/192.0.0.0/ipcidr/24",
                "/ip4/192.0.2.0/ipcidr/24",
                "/ip4/192.168.0.0/ipcidr/16",
                "/ip4/198.18.0.0/ipcidr/15",
                "/ip4/198.51.100.0/ipcidr/24",
                "/ip4/203.0.113.0/ipcidr/24",
                "/ip4/240.0.0.0/ipcidr/4",
                "/ip6/100::/ipcidr/64",
                "/ip6/2001:2::/ipcidr/48",
                "/ip6/2001:db8::/ipcidr/32",
                "/ip6/fc00::/ipcidr/7",
                "/ip6/fe80::/ipcidr/10",
        }
        conf.Addresses.NoAnnounce = conf.Swarm.AddrFilters

        if enableLocalDiscovery {
                conf.Discovery.MDNS.Enabled = true
                conf.Swarm.DisableNatPortMap = false
                conf.Swarm.AddrFilters = []string{}
                conf.Addresses.NoAnnounce = []string{}
        }

        if err := fsrepo.Init(repoRoot, conf); err != nil {
                return nil, err
        }
        return conf, nil
}

func checkWritable(dir string) error {
        _, err := os.Stat(dir)
        if err == nil {
                // dir exists, make sure we can write to it
                testfile := path.Join(dir, "test")
                fi, err := os.Create(testfile)
                if err != nil {
                        if os.IsPermission(err) {
                                return fmt.Errorf("%s is not writeable by the current user", dir)
                        }
                        return fmt.Errorf("unexpected error while checking writeablility of repo root: %s", err)
                }
                if err := fi.Close(); err != nil {
                        return err
                }
                return os.Remove(testfile)
        }

        if os.IsNotExist(err) {
                // dir doesn't exist, check that we can create it
                return os.Mkdir(dir, 0750)
        }

        if os.IsPermission(err) {
                return fmt.Errorf("cannot write to %s, incorrect permissions: %w", dir, err)
        }

        return err
}

1	package ipfs
2
3	import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"os"
9	"path"
10	"sync"
11
12	coreiface "github.com/ipfs/boxo/coreiface"
13	ihelper "github.com/ipfs/boxo/ipld/unixfs/importer/helpers"
14	ipfscid "github.com/ipfs/go-cid"
15	"github.com/ipfs/kubo/commands"
16	"github.com/ipfs/kubo/config"
17	ipfscore "github.com/ipfs/kubo/core"
18	ipfsapi "github.com/ipfs/kubo/core/coreapi"
19	"github.com/ipfs/kubo/plugin/loader"
20	"github.com/ipfs/kubo/repo"
21	"github.com/ipfs/kubo/repo/fsrepo"
22	"github.com/ipfs/kubo/repo/fsrepo/migrations"
23	"github.com/ipfs/kubo/repo/fsrepo/migrations/ipfsfetcher"
24	"github.com/multiformats/go-multicodec"
25	"github.com/multiformats/go-multihash"
26
27	"go.vocdoni.io/dvote/log"
28	)
29
30	var ConfigRoot string
31
32	// ChunkerTypeSize is the chunker type used by IPFS to calculate to build the DAG.
33	const ChunkerTypeSize = "size-262144"
34
35	func init() {	29✔
36	// Initialize the DAG builder with offline exchange and the correct CID format	29✔
37	format := ipfscid.V1Builder{	29✔
38	Codec: uint64(multicodec.DagJson),	29✔
39	MhType: uint64(multihash.SHA2_256),	29✔
40	}	29✔
41	dAGbuilder = ihelper.DagBuilderParams{	29✔
42	Dagserv: dAG(),	29✔
43	RawLeaves: false,	29✔
44	Maxlinks: ihelper.DefaultLinksPerBlock,	29✔
45	NoCopy: false,	29✔
46	CidBuilder: &format,	29✔
47	}	29✔
48	}	29✔
49
50	// Init initializes the IPFS node and repository.
51	func initRepository(enableLocalDiscovery bool) error {	2✔
52	daemonLocked, err := fsrepo.LockedByOtherProcess(ConfigRoot)	2✔
53	if err != nil {	2✔
54	return err	×
55	}	×
56	log.Info("checking if daemon is running")	2✔
57	if daemonLocked {	2✔
58	log.Debug("ipfs daemon is running")	×
59	return fmt.Errorf("ipfs daemon is running. please stop it to run this command")	×
60	}	×
61
62	if err := os.MkdirAll(ConfigRoot, 0750); err != nil {	2✔
63	return err	×
64	}	×
65
66	if err := installDatabasePlugins(); err != nil {	2✔
67	return err	×
68	}	×
69	_, err = doInit(io.Discard, ConfigRoot, 2048, enableLocalDiscovery)	2✔
70	return err	2✔
71	}
72
73	// StartNode starts the IPFS node.
74	func (i Handler) startNode() (ipfscore.IpfsNode, coreiface.CoreAPI, error) {	2✔
75	log.Infow("starting IPFS node", "config", ConfigRoot)	2✔
76		2✔
77	// check if needs init	2✔
78	if !fsrepo.IsInitialized(ConfigRoot) {	4✔
79	if err := initRepository(i.EnableLocalDiscovery); err != nil {	2✔
NEW 80	return nil, nil, err	×
NEW 81	}	×
82	}
83
84	r, err := fsrepo.Open(ConfigRoot)	2✔
85	if errors.Is(err, fsrepo.ErrNeedMigration) {	2✔
86	log.Warn("Found outdated ipfs repo, migrations need to be run.")	×
87	r, err = runMigrationsAndOpen(ConfigRoot)	×
88	}	×
89	if err != nil {	2✔
90	return nil, nil, err	×
91	}	×
92	cfg := &ipfscore.BuildCfg{	2✔
93	Repo: r,	2✔
94	Online: true,	2✔
95	Permanent: true,	2✔
96	}	2✔
97		2✔
98	// We use node.Cancel to stop it instead.	2✔
99	ctx := context.Background()	2✔
100		2✔
101	node, err := ipfscore.NewNode(ctx, cfg)	2✔
102	if err != nil {	2✔
103	return nil, nil, err	×
104	}	×
105	node.IsDaemon = true	2✔
106	node.IsOnline = true	2✔
107		2✔
108	api, err := ipfsapi.NewCoreAPI(node)	2✔
109	if err != nil {	2✔
110	return nil, nil, err	×
111	}	×
112	return node, api, nil	2✔
113	}
114
115	// runMigrationsAndOpen fetches and applies migrations just like upstream kubo does
116	// and returns fsrepo.Open(ConfigRoot)
117	func runMigrationsAndOpen(ConfigRoot string) (repo.Repo, error) {	×
118	// Read Migration section of IPFS config	×
119	migrationCfg, err := migrations.ReadMigrationConfig(ConfigRoot, "")	×
120	if err != nil {	×
121	return nil, err	×
122	}	×
123
124	// Define function to create IPFS fetcher. Do not supply an
125	// already-constructed IPFS fetcher, because this may be expensive and
126	// not needed according to migration config. Instead, supply a function
127	// to construct the particular IPFS fetcher implementation used here,
128	// which is called only if an IPFS fetcher is needed.
129	newIpfsFetcher := func(distPath string) migrations.Fetcher {	×
130	return ipfsfetcher.NewIpfsFetcher(distPath, 0, &ConfigRoot, "")	×
131	}	×
132
133	// Fetch migrations from current distribution, or location from environ
134	fetchDistPath := migrations.GetDistPathEnv(migrations.CurrentIpfsDist)	×
135		×
136	// Create fetchers according to migrationCfg.DownloadSources	×
137	fetcher, err := migrations.GetMigrationFetcher(migrationCfg.DownloadSources,	×
138	fetchDistPath, newIpfsFetcher)	×
139	if err != nil {	×
140	return nil, err	×
141	}	×
142	defer fetcher.Close()	×
143		×
144	if migrationCfg.Keep == "cache" \|\| migrationCfg.Keep == "pin" {	×
145	// Create temp directory to store downloaded migration archives	×
146	migrations.DownloadDirectory, err = os.MkdirTemp("", "migrations")	×
147	if err != nil {	×
148	return nil, err	×
149	}	×
150	// Defer cleanup of download directory so that it gets cleaned up
151	// if daemon returns early due to error
152	defer func() {	×
153	if migrations.DownloadDirectory != "" {	×
154	_ = os.RemoveAll(migrations.DownloadDirectory)	×
155	}	×
156	}()
157	}
158
159	err = migrations.RunMigration(context.TODO(), fetcher, fsrepo.RepoVersion, ConfigRoot, false)	×
160	if err != nil {	×
161	return nil, fmt.Errorf("migrations of ipfs-repo failed: %w", err)	×
162	}	×
163
164	return fsrepo.Open(ConfigRoot)	×
165	}
166
167	// CmdCtx returns a commands.Context for the given node and repo path.
168	func cmdCtx(node *ipfscore.IpfsNode, repoPath string) commands.Context {	2✔
169	return commands.Context{	2✔
170	ConfigRoot: repoPath,	2✔
171	ConstructNode: func() (*ipfscore.IpfsNode, error) {	2✔
172	return node, nil	×
173	},	×
174	}
175	}
176
177	var installDatabasePlugins = sync.OnceValue(func() error {	2✔
178	loader, err := loader.NewPluginLoader("")	2✔
179	if err != nil {	2✔
180	return err	×
181	}	×
182	if err := loader.Initialize(); err != nil {	2✔
183	return err	×
184	}	×
185	if err := loader.Inject(); err != nil {	2✔
186	return err	×
187	}	×
188	return nil	2✔
189	})
190
191	func doInit(out io.Writer, repoRoot string, nBitsForKeypair int, enableLocalDiscovery bool) (*config.Config, error) {	2✔
192	log.Infow("initializing new IPFS repository", "root", repoRoot)	2✔
193	if err := checkWritable(repoRoot); err != nil {	2✔
194	return nil, err	×
195	}	×
196
197	if fsrepo.IsInitialized(repoRoot) {	2✔
198	return nil, fmt.Errorf("repo exists")	×
199	}	×
200
201	conf, err := config.Init(out, nBitsForKeypair)	2✔
202	if err != nil {	2✔
203	return nil, err	×
204	}	×
205
206	// Apply `server` configuration profile:
207	// Disables local host discovery, recommended when running IPFS on machines with public IPv4 addresses
208	// Prevent from scanning local networks which can trigger netscan alerts.
209	// See: https://github.com/ipfs/kubo/issues/7985
210	conf.Discovery.MDNS.Enabled = false	2✔
211	conf.Swarm.DisableNatPortMap = true	2✔
212	conf.Swarm.AddrFilters = []string{	2✔
213	"/ip4/10.0.0.0/ipcidr/8",	2✔
214	"/ip4/100.64.0.0/ipcidr/10",	2✔
215	"/ip4/169.254.0.0/ipcidr/16",	2✔
216	"/ip4/172.16.0.0/ipcidr/12",	2✔
217	"/ip4/192.0.0.0/ipcidr/24",	2✔
218	"/ip4/192.0.2.0/ipcidr/24",	2✔
219	"/ip4/192.168.0.0/ipcidr/16",	2✔
220	"/ip4/198.18.0.0/ipcidr/15",	2✔
221	"/ip4/198.51.100.0/ipcidr/24",	2✔
222	"/ip4/203.0.113.0/ipcidr/24",	2✔
223	"/ip4/240.0.0.0/ipcidr/4",	2✔
224	"/ip6/100::/ipcidr/64",	2✔
225	"/ip6/2001:2::/ipcidr/48",	2✔
226	"/ip6/2001:db8::/ipcidr/32",	2✔
227	"/ip6/fc00::/ipcidr/7",	2✔
228	"/ip6/fe80::/ipcidr/10",	2✔
229	}	2✔
230	conf.Addresses.NoAnnounce = conf.Swarm.AddrFilters	2✔
231		2✔
232	if enableLocalDiscovery {	3✔
233	conf.Discovery.MDNS.Enabled = true	1✔
234	conf.Swarm.DisableNatPortMap = false	1✔
235	conf.Swarm.AddrFilters = []string{}	1✔
236	conf.Addresses.NoAnnounce = []string{}	1✔
237	}	1✔
238
239	if err := fsrepo.Init(repoRoot, conf); err != nil {	2✔
240	return nil, err	×
241	}	×
242	return conf, nil	2✔
243	}
244
245	func checkWritable(dir string) error {	2✔
246	_, err := os.Stat(dir)	2✔
247	if err == nil {	4✔
248	// dir exists, make sure we can write to it	2✔
249	testfile := path.Join(dir, "test")	2✔
250	fi, err := os.Create(testfile)	2✔
251	if err != nil {	2✔
252	if os.IsPermission(err) {	×
253	return fmt.Errorf("%s is not writeable by the current user", dir)	×
254	}	×
255	return fmt.Errorf("unexpected error while checking writeablility of repo root: %s", err)	×
256	}
257	if err := fi.Close(); err != nil {	2✔
258	return err	×
259	}	×
260	return os.Remove(testfile)	2✔
261	}
262
263	if os.IsNotExist(err) {	×
264	// dir doesn't exist, check that we can create it	×
265	return os.Mkdir(dir, 0750)	×
266	}	×
267
268	if os.IsPermission(err) {	×
269	return fmt.Errorf("cannot write to %s, incorrect permissions: %w", dir, err)	×
270	}	×
271
272	return err	×
273	}

vocdoni / vocdoni-node / 6863543637

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous