cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge pull request #506 from filecoin-project/feat/dual-config

Browse Source 
Add support for different configs
This commit is contained in:
Łukasz Magiera 2019-10-31 09:02:10 +01:00 committed by GitHub
commit 5c065f0187
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 219 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
chain/gen/gen.go
View File

@ -75,7 +75,7 @@ func (m mybs) Get(c cid.Cid) (block.Block, error) {
func NewGenerator() (*ChainGen, error) {
mr := repo.NewMemory(nil)
lr, err := mr.Lock()
lr, err := mr.Lock(repo.RepoStorageMiner)
if err != nil {
return nil, xerrors.Errorf("taking mem-repo lock failed: %w", err)
}

4
cmd/lotus-storage-miner/init.go
View File

@ -101,7 +101,7 @@ var initCmd = &cli.Command{
log.Info("Initializing repo")
if err := r.Init(); err != nil {
if err := r.Init(repo.RepoStorageMiner); err != nil {
return err
}
@ -122,7 +122,7 @@ var initCmd = &cli.Command{
}
func storageMinerInit(ctx context.Context, cctx *cli.Context, api api.FullNode, r repo.Repo) error {
lr, err := r.Lock()
lr, err := r.Lock(repo.RepoStorageMiner)
if err != nil {
return err
}

5
cmd/lotus/daemon.go
View File

@ -4,9 +4,10 @@ package main
import (
"context"
"github.com/filecoin-project/lotus/peermgr"
"io/ioutil"
"github.com/filecoin-project/lotus/peermgr"
"github.com/multiformats/go-multiaddr"
"golang.org/x/xerrors"
"gopkg.in/urfave/cli.v2"
@ -53,7 +54,7 @@ var DaemonCmd = &cli.Command{
return err
}
if err := r.Init(); err != nil && err != repo.ErrRepoExists {
if err := r.Init(repo.RepoFullNode); err != nil && err != repo.ErrRepoExists {
return err
}

117
node/builder.go
View File

@ -15,6 +15,7 @@ import (
pubsub "github.com/libp2p/go-libp2p-pubsub"
record "github.com/libp2p/go-libp2p-record"
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain"
@ -100,11 +101,6 @@ const (
_nInvokes // keep this last
)
const (
nodeFull = iota
nodeStorageMiner
)
type Settings struct {
// modules is a map of constructors for DI
//
@ -117,14 +113,12 @@ type Settings struct {
// type, and must be applied in correct order
invokes []fx.Option
nodeType int
nodeType repo.RepoType
Online bool // Online option applied
Config bool // Config option applied
}
var defConf = config.Default()
func defaults() []Option {
return []Option{
Override(new(helpers.MetricsCtx), context.Background),
@ -161,10 +155,14 @@ func libp2p() Option {
Override(new(*pubsub.PubSub), lp2p.GossipSub()),
Override(PstoreAddSelfKeysKey, lp2p.PstoreAddSelfKeys),
Override(StartListeningKey, lp2p.StartListening(defConf.Libp2p.ListenAddresses)),
Override(StartListeningKey, lp2p.StartListening(config.DefaultFullNode().Libp2p.ListenAddresses)),
)
}
func isType(t repo.RepoType) func(s *Settings) bool {
return func(s *Settings) bool { return s.nodeType == t }
}
// Online sets up basic libp2p node
func Online() Option {
return Options(
@ -181,7 +179,7 @@ func Online() Option {
// Full node
ApplyIf(func(s *Settings) bool { return s.nodeType == nodeFull },
ApplyIf(isType(repo.RepoFullNode),
// TODO: Fix offline mode
Override(new(dtypes.BootstrapPeers), modules.BuiltinBootstrap),
@ -230,7 +228,7 @@ func Online() Option {
),
// Storage miner
ApplyIf(func(s *Settings) bool { return s.nodeType == nodeStorageMiner },
ApplyIf(func(s *Settings) bool { return s.nodeType == repo.RepoStorageMiner },
Override(new(*sectorbuilder.SectorBuilder), sectorbuilder.New),
Override(new(*sector.Store), sector.NewStore),
Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
@ -260,7 +258,7 @@ func StorageMiner(out *api.StorageMiner) Option {
),
func(s *Settings) error {
s.nodeType = nodeStorageMiner
s.nodeType = repo.RepoStorageMiner
return nil
},
@ -274,7 +272,7 @@ func StorageMiner(out *api.StorageMiner) Option {
}
// Config sets up constructors based on the provided Config
func Config(cfg *config.Root) Option {
func ConfigCommon(cfg *config.Common) Option {
return Options(
func(s *Settings) error { s.Config = true; return nil },
@ -284,45 +282,71 @@ func Config(cfg *config.Root) Option {
ApplyIf(func(s *Settings) bool { return len(cfg.Libp2p.BootstrapPeers) > 0 },
Override(new(dtypes.BootstrapPeers), modules.ConfigBootstrap(cfg.Libp2p.BootstrapPeers)),
),
ApplyIf(func(s *Settings) bool { return s.nodeType == nodeFull },
Override(HeadMetricsKey, metrics.SendHeadNotifs(cfg.Metrics.Nickname)),
),
),
)
}
func Repo(r repo.Repo) Option {
lr, err := r.Lock()
if err != nil {
return Error(err)
}
cfg, err := lr.Config()
if err != nil {
return Error(err)
}
func ConfigFullNode(cfg *config.FullNode) Option {
//ApplyIf(func(s *Settings) bool { return s.nodeType == repo.RepoFullNode }),
return Options(
Config(cfg),
Override(new(repo.LockedRepo), modules.LockedRepo(lr)), // module handles closing
Override(new(dtypes.MetadataDS), modules.Datastore),
Override(new(dtypes.ChainBlockstore), modules.ChainBlockstore),
Override(new(dtypes.ClientFilestore), modules.ClientFstore),
Override(new(dtypes.ClientBlockstore), modules.ClientBlockstore),
Override(new(dtypes.ClientDAG), modules.ClientDAG),
Override(new(ci.PrivKey), lp2p.PrivKey),
Override(new(ci.PubKey), ci.PrivKey.GetPublic),
Override(new(peer.ID), peer.IDFromPublicKey),
Override(new(types.KeyStore), modules.KeyStore),
Override(new(*dtypes.APIAlg), modules.APISecret),
ConfigCommon(&cfg.Common),
Override(HeadMetricsKey, metrics.SendHeadNotifs(cfg.Metrics.Nickname)),
)
}
func configFull(c interface{}) Option {
cfg, ok := c.(*config.FullNode)
if !ok {
return Error(xerrors.Errorf("invalid config from repo, got: %T", c))
}
return ConfigFullNode(cfg)
}
func configMiner(c interface{}) Option {
cfg, ok := c.(*config.StorageMiner)
if !ok {
return Error(xerrors.Errorf("invalid config from repo, got: %T", c))
}
return ConfigCommon(&cfg.Common)
}
func Repo(r repo.Repo) Option {
return func(settings *Settings) error {
lr, err := r.Lock(settings.nodeType)
if err != nil {
return err
}
c, err := lr.Config()
if err != nil {
return err
}
return Options(
Override(new(repo.LockedRepo), modules.LockedRepo(lr)), // module handles closing
ApplyIf(isType(repo.RepoFullNode), configFull(c)),
ApplyIf(isType(repo.RepoStorageMiner), configMiner(c)),
Override(new(dtypes.MetadataDS), modules.Datastore),
Override(new(dtypes.ChainBlockstore), modules.ChainBlockstore),
Override(new(dtypes.ClientFilestore), modules.ClientFstore),
Override(new(dtypes.ClientBlockstore), modules.ClientBlockstore),
Override(new(dtypes.ClientDAG), modules.ClientDAG),
Override(new(ci.PrivKey), lp2p.PrivKey),
Override(new(ci.PubKey), ci.PrivKey.GetPublic),
Override(new(peer.ID), peer.IDFromPublicKey),
Override(new(types.KeyStore), modules.KeyStore),
Override(new(*dtypes.APIAlg), modules.APISecret),
)(settings)
}
}
func FullAPI(out *api.FullNode) Option {
return func(s *Settings) error {
resAPI := &impl.FullNodeAPI{}
@ -337,8 +361,9 @@ type StopFunc func(context.Context) error
// New builds and starts new Filecoin node
func New(ctx context.Context, opts ...Option) (StopFunc, error) {
settings := Settings{
modules: map[interface{}]fx.Option{},
invokes: make([]fx.Option, _nInvokes),
modules: map[interface{}]fx.Option{},
invokes: make([]fx.Option, _nInvokes),
nodeType: repo.RepoFullNode,
}
// apply module options in the right order

49
node/config/def.go
View File

@ -1,15 +1,27 @@
package config
import "time"
import (
"encoding"
"time"
)
// Root is starting point of the config
type Root struct {
// Common is common config between full node and miner
type Common struct {
API API
Libp2p Libp2p
}
// FullNode is a full node config
type FullNode struct {
Common
Metrics Metrics
}
// StorageMiner is a storage miner config
type StorageMiner struct {
Common
}
// API contains configs for API endpoint
type API struct {
ListenAddress string
@ -26,9 +38,8 @@ type Metrics struct {
Nickname string
}
// Default returns the default config
func Default() *Root {
def := Root{
func defCommon() Common {
return Common{
API: API{
ListenAddress: "/ip6/::1/tcp/1234/http",
Timeout: Duration(30 * time.Second),
@ -40,10 +51,27 @@ func Default() *Root {
},
},
}
return &def
}
// Duration is a wrapper type for time.Duration for decoding it from TOML
// Default returns the default config
func DefaultFullNode() *FullNode {
return &FullNode{
Common: defCommon(),
}
}
func DefaultStorageMiner() *StorageMiner {
return &StorageMiner{
Common: defCommon(),
}
}
var _ encoding.TextMarshaler = (*Duration)(nil)
var _ encoding.TextUnmarshaler = (*Duration)(nil)
// Duration is a wrapper type for time.Duration
// for decoding and encoding from/to TOML
type Duration time.Duration
// UnmarshalText implements interface for TOML decoding
@ -55,3 +83,8 @@ func (dur *Duration) UnmarshalText(text []byte) error {
*dur = Duration(d)
return err
}
func (dur Duration) MarshalText() ([]byte, error) {
d := time.Duration(dur)
return []byte(d.String()), nil
}

27
node/config/load.go
View File

@ -1,30 +1,32 @@
package config
import (
"bytes"
"io"
"os"
"github.com/BurntSushi/toml"
"golang.org/x/xerrors"
)
// FromFile loads config from a specified file overriding defaults specified in
// the source code. If file does not exist or is empty defaults are asummed.
func FromFile(path string) (*Root, error) {
// the def parameter. If file does not exist or is empty defaults are asummed.
func FromFile(path string, def interface{}) (interface{}, error) {
file, err := os.Open(path)
switch {
case os.IsNotExist(err):
return Default(), nil
return def, nil
case err != nil:
return nil, err
}
defer file.Close() //nolint:errcheck // The file is RO
return FromReader(file)
return FromReader(file, def)
}
// FromReader loads config from a reader instance.
func FromReader(reader io.Reader) (*Root, error) {
cfg := Default()
func FromReader(reader io.Reader, def interface{}) (interface{}, error) {
cfg := def
_, err := toml.DecodeReader(reader, cfg)
if err != nil {
return nil, err
@ -32,3 +34,16 @@ func FromReader(reader io.Reader) (*Root, error) {
return cfg, nil
}
func ConfigComment(t interface{}) ([]byte, error) {
buf := new(bytes.Buffer)
_, _ = buf.WriteString("# Default config:\n")
e := toml.NewEncoder(buf)
if err := e.Encode(t); err != nil {
return nil, xerrors.Errorf("encoding config: %w", err)
}
b := buf.Bytes()
b = bytes.ReplaceAll(b, []byte("\n"), []byte("\n#"))
return b, nil
}

14
node/config/load_test.go
View File

@ -14,16 +14,16 @@ func TestDecodeNothing(t *testing.T) {
assert := assert.New(t)
{
cfg, err := FromFile(os.DevNull)
cfg, err := FromFile(os.DevNull, DefaultFullNode())
assert.Nil(err, "error should be nil")
assert.Equal(Default(), cfg,
assert.Equal(DefaultFullNode(), cfg,
"config from empty file should be the same as default")
}
{
cfg, err := FromFile("./does-not-exist.toml")
cfg, err := FromFile("./does-not-exist.toml", DefaultFullNode())
assert.Nil(err, "error should be nil")
assert.Equal(Default(), cfg,
assert.Equal(DefaultFullNode(), cfg,
"config from not exisiting file should be the same as default")
}
}
@ -34,11 +34,11 @@ func TestParitalConfig(t *testing.T) {
[API]
Timeout = "10s"
`
expected := Default()
expected := DefaultFullNode()
expected.API.Timeout = Duration(10 * time.Second)
{
cfg, err := FromReader(bytes.NewReader([]byte(cfgString)))
cfg, err := FromReader(bytes.NewReader([]byte(cfgString)), DefaultFullNode())
assert.NoError(err, "error should be nil")
assert.Equal(expected, cfg,
"config from reader should contain changes")
@ -55,7 +55,7 @@ func TestParitalConfig(t *testing.T) {
assert.NoError(err, "closing tmp file should not error")
defer os.Remove(fname) //nolint:errcheck
cfg, err := FromFile(fname)
cfg, err := FromFile(fname, DefaultFullNode())
assert.Nil(err, "error should be nil")
assert.Equal(expected, cfg,
"config from reader should contain changes")

2
node/node_test.go
View File

@ -34,7 +34,7 @@ import (
func testStorageNode(ctx context.Context, t *testing.T, waddr address.Address, act address.Address, tnd test.TestNode) test.TestStorageNode {
r := repo.NewMemory(nil)
lr, err := r.Lock()
lr, err := r.Lock(repo.RepoStorageMiner)
require.NoError(t, err)
pk, _, err := crypto.GenerateEd25519Key(rand.Reader)

71
node/repo/fsrepo.go
View File

@ -2,6 +2,7 @@ package repo
import (
"encoding/json"
"fmt"
"io"
"io/ioutil"
"os"
@ -33,13 +34,33 @@ const (
fsKeystore = "keystore"
)
type RepoType int
const (
_ = iota // Default is invalid
RepoFullNode RepoType = iota
RepoStorageMiner
)
func defConfForType(t RepoType) interface{} {
switch t {
case RepoFullNode:
return config.DefaultFullNode()
case RepoStorageMiner:
return config.DefaultStorageMiner()
default:
panic(fmt.Sprintf("unknown RepoType(%d)", int(t)))
}
}
var log = logging.Logger("repo")
var ErrRepoExists = xerrors.New("repo exists")
// FsRepo is struct for repo, use NewFS to create
type FsRepo struct {
path string
path string
repoType RepoType
}
var _ Repo = &FsRepo{}
@ -65,7 +86,7 @@ func (fsr *FsRepo) Exists() (bool, error) {
return !notexist, err
}
func (fsr *FsRepo) Init() error {
func (fsr *FsRepo) Init(t RepoType) error {
exist, err := fsr.Exists()
if err != nil {
return err
@ -79,18 +100,36 @@ func (fsr *FsRepo) Init() error {
if err != nil && !os.IsExist(err) {
return err
}
c, err := os.Create(filepath.Join(fsr.path, fsConfig))
if err != nil {
return err
}
if err := c.Close(); err != nil {
return err
if err := fsr.initConfig(t); err != nil {
return xerrors.Errorf("init config: %w", err)
}
return fsr.initKeystore()
}
func (fsr *FsRepo) initConfig(t RepoType) error {
c, err := os.Create(filepath.Join(fsr.path, fsConfig))
if err != nil {
return err
}
comm, err := config.ConfigComment(defConfForType(t))
if err != nil {
return xerrors.Errorf("comment: %w", err)
}
_, err = c.Write(comm)
if err != nil {
return xerrors.Errorf("write config: %w", err)
}
if err := c.Close(); err != nil {
return xerrors.Errorf("close config: %w", err)
}
return nil
}
func (fsr *FsRepo) initKeystore() error {
kstorePath := filepath.Join(fsr.path, fsKeystore)
if _, err := os.Stat(kstorePath); err == nil {
@ -142,7 +181,7 @@ func (fsr *FsRepo) APIToken() ([]byte, error) {
}
// Lock acquires exclusive lock on this repo
func (fsr *FsRepo) Lock() (LockedRepo, error) {
func (fsr *FsRepo) Lock(repoType RepoType) (LockedRepo, error) {
locked, err := fslock.Locked(fsr.path, fsLock)
if err != nil {
return nil, xerrors.Errorf("could not check lock status: %w", err)
@ -156,14 +195,16 @@ func (fsr *FsRepo) Lock() (LockedRepo, error) {
return nil, xerrors.Errorf("could not lock the repo: %w", err)
}
return &fsLockedRepo{
path: fsr.path,
closer: closer,
path: fsr.path,
repoType: repoType,
closer: closer,
}, nil
}
type fsLockedRepo struct {
path string
closer io.Closer
path string
repoType RepoType
closer io.Closer
ds datastore.Batching
dsErr error
@ -219,11 +260,11 @@ func (fsr *fsLockedRepo) Datastore(ns string) (datastore.Batching, error) {
return namespace.Wrap(fsr.ds, datastore.NewKey(ns)), nil
}
func (fsr *fsLockedRepo) Config() (*config.Root, error) {
func (fsr *fsLockedRepo) Config() (interface{}, error) {
if err := fsr.stillValid(); err != nil {
return nil, err
}
return config.FromFile(fsr.join(fsConfig))
return config.FromFile(fsr.join(fsConfig), defConfForType(fsr.repoType))
}
func (fsr *fsLockedRepo) SetAPIEndpoint(ma multiaddr.Multiaddr) error {

2
node/repo/fsrepo_test.go
View File

@ -17,7 +17,7 @@ func genFsRepo(t *testing.T) (*FsRepo, func()) {
t.Fatal(err)
}
err = repo.Init()
err = repo.Init(RepoFullNode)
if err != ErrRepoExists && err != nil {
t.Fatal(err)
}

5
node/repo/interface.go
View File

@ -7,7 +7,6 @@ import (
"github.com/multiformats/go-multiaddr"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/config"
)
var (
@ -25,7 +24,7 @@ type Repo interface {
APIToken() ([]byte, error)
// Lock locks the repo for exclusive use.
Lock() (LockedRepo, error)
Lock(RepoType) (LockedRepo, error)
}
type LockedRepo interface {
@ -36,7 +35,7 @@ type LockedRepo interface {
Datastore(namespace string) (datastore.Batching, error)
// Returns config in this repo
Config() (*config.Root, error)
Config() (interface{}, error)
// SetAPIEndpoint sets the endpoint of the current API
// so it can be read by API clients

15
node/repo/memrepo.go
View File

@ -10,7 +10,6 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/config"
)
type MemRepo struct {
@ -24,12 +23,13 @@ type MemRepo struct {
token *byte
datastore datastore.Datastore
configF func() *config.Root
configF func(t RepoType) interface{}
keystore map[string]types.KeyInfo
}
type lockedMemRepo struct {
mem *MemRepo
t RepoType
sync.RWMutex
token *byte
@ -44,7 +44,7 @@ var _ Repo = &MemRepo{}
// MemRepoOptions contains options for memory repo
type MemRepoOptions struct {
Ds datastore.Datastore
ConfigF func() *config.Root
ConfigF func(RepoType) interface{}
KeyStore map[string]types.KeyInfo
}
@ -56,7 +56,7 @@ func NewMemory(opts *MemRepoOptions) *MemRepo {
opts = &MemRepoOptions{}
}
if opts.ConfigF == nil {
opts.ConfigF = config.Default
opts.ConfigF = defConfForType
}
if opts.Ds == nil {
opts.Ds = dssync.MutexWrap(datastore.NewMapDatastore())
@ -92,7 +92,7 @@ func (mem *MemRepo) APIToken() ([]byte, error) {
return mem.api.token, nil
}
func (mem *MemRepo) Lock() (LockedRepo, error) {
func (mem *MemRepo) Lock(t RepoType) (LockedRepo, error) {
select {
case mem.repoLock <- struct{}{}:
default:
@ -102,6 +102,7 @@ func (mem *MemRepo) Lock() (LockedRepo, error) {
return &lockedMemRepo{
mem: mem,
t: t,
token: mem.token,
}, nil
}
@ -143,11 +144,11 @@ func (lmem *lockedMemRepo) Datastore(ns string) (datastore.Batching, error) {
return namespace.Wrap(lmem.mem.datastore, datastore.NewKey(ns)), nil
}
func (lmem *lockedMemRepo) Config() (*config.Root, error) {
func (lmem *lockedMemRepo) Config() (interface{}, error) {
if err := lmem.checkToken(); err != nil {
return nil, err
}
return lmem.mem.configF(), nil
return lmem.mem.configF(lmem.t), nil
}
func (lmem *lockedMemRepo) SetAPIEndpoint(ma multiaddr.Multiaddr) error {

10
node/repo/repo_test.go
View File

@ -18,12 +18,12 @@ func basicTest(t *testing.T, repo Repo) {
}
assert.Nil(t, apima, "with no api endpoint, return should be nil")
lrepo, err := repo.Lock()
lrepo, err := repo.Lock(RepoFullNode)
assert.NoError(t, err, "should be able to lock once")
assert.NotNil(t, lrepo, "locked repo shouldn't be nil")
{
lrepo2, err := repo.Lock()
lrepo2, err := repo.Lock(RepoFullNode)
if assert.Error(t, err) {
assert.Equal(t, ErrRepoAlreadyLocked, err)
}
@ -33,7 +33,7 @@ func basicTest(t *testing.T, repo Repo) {
err = lrepo.Close()
assert.NoError(t, err, "should be able to unlock")
lrepo, err = repo.Lock()
lrepo, err = repo.Lock(RepoFullNode)
assert.NoError(t, err, "should be able to relock")
assert.NotNil(t, lrepo, "locked repo shouldn't be nil")
@ -48,7 +48,7 @@ func basicTest(t *testing.T, repo Repo) {
assert.Equal(t, ma, apima, "returned API multiaddr should be the same")
cfg, err := lrepo.Config()
assert.Equal(t, config.Default(), cfg, "there should be a default config")
assert.Equal(t, config.DefaultFullNode(), cfg, "there should be a default config")
assert.NoError(t, err, "config should not error")
err = lrepo.Close()
@ -64,7 +64,7 @@ func basicTest(t *testing.T, repo Repo) {
k1 := types.KeyInfo{Type: "foo"}
k2 := types.KeyInfo{Type: "bar"}
lrepo, err = repo.Lock()
lrepo, err = repo.Lock(RepoFullNode)
assert.NoError(t, err, "should be able to relock")
assert.NotNil(t, lrepo, "locked repo shouldn't be nil")