frostfs-node/pkg/innerring/innerring.go

572 lines
15 KiB
Go
Raw Normal View History

2020-07-24 13:54:03 +00:00
package innerring
import (
"context"
"crypto/ecdsa"
"github.com/nspcc-dev/neo-go/pkg/core/block"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
"github.com/nspcc-dev/neo-go/pkg/encoding/fixedn"
2020-07-24 13:54:03 +00:00
"github.com/nspcc-dev/neo-go/pkg/util"
crypto "github.com/nspcc-dev/neofs-crypto"
"github.com/nspcc-dev/neofs-node/pkg/innerring/invoke"
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/alphabet"
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/audit"
2020-07-24 13:54:03 +00:00
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/balance"
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/container"
2020-07-24 13:54:03 +00:00
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/neofs"
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/netmap"
"github.com/nspcc-dev/neofs-node/pkg/innerring/timers"
"github.com/nspcc-dev/neofs-node/pkg/morph/client"
auditWrapper "github.com/nspcc-dev/neofs-node/pkg/morph/client/audit/wrapper"
2020-07-24 13:54:03 +00:00
"github.com/nspcc-dev/neofs-node/pkg/morph/event"
"github.com/nspcc-dev/neofs-node/pkg/morph/subscriber"
audittask "github.com/nspcc-dev/neofs-node/pkg/services/audit/taskmanager"
util2 "github.com/nspcc-dev/neofs-node/pkg/util"
"github.com/nspcc-dev/neofs-node/pkg/util/precision"
"github.com/panjf2000/ants/v2"
2020-07-24 13:54:03 +00:00
"github.com/pkg/errors"
"github.com/spf13/viper"
"go.uber.org/atomic"
"go.uber.org/zap"
)
type (
// Server is the inner ring application structure, that contains all event
// processors, shared variables and event handlers.
Server struct {
log *zap.Logger
// event producers
morphListener event.Listener
mainnetListener event.Listener
blockTimers []*timers.BlockTimer
epochTimer *timers.BlockTimer
2020-07-24 13:54:03 +00:00
// global state
morphClient *client.Client
mainnetClient *client.Client
epochCounter atomic.Uint64
innerRingIndex atomic.Int32
innerRingSize atomic.Int32
precision precision.Fixed8Converter
auditClient *auditWrapper.ClientWrapper
2020-07-24 13:54:03 +00:00
// internal variables
key *ecdsa.PrivateKey
pubKey []byte
contracts *contracts
predefinedValidators []keys.PublicKey
workers []func(context.Context)
2020-07-24 13:54:03 +00:00
}
contracts struct {
neofs util.Uint160 // in mainnet
netmap util.Uint160 // in morph
balance util.Uint160 // in morph
container util.Uint160 // in morph
audit util.Uint160 // in morph
gas util.Uint160 // native contract in both chains
alphabet [alphabetContractsN]util.Uint160 // in morph
2020-07-24 13:54:03 +00:00
}
chainParams struct {
log *zap.Logger
cfg *viper.Viper
key *ecdsa.PrivateKey
name string
gas util.Uint160
}
)
const (
morphPrefix = "morph"
mainnetPrefix = "mainnet"
alphabetContractsN = 7 // az, buky, vedi, glagoli, dobro, jest, zhivete
2020-07-24 13:54:03 +00:00
)
// Start runs all event providers.
func (s *Server) Start(ctx context.Context, intError chan<- error) error {
err := s.initConfigFromBlockchain()
if err != nil {
return err
}
// vote for sidechain validator if it is prepared in config
err = s.voteForSidechainValidator(s.predefinedValidators)
if err != nil {
// we don't stop inner ring execution on this error
s.log.Warn("can't vote for prepared validators",
zap.String("error", err.Error()))
}
morphErr := make(chan error)
mainnnetErr := make(chan error)
// anonymous function to multiplex error channels
go func() {
select {
case <-ctx.Done():
return
case err := <-morphErr:
intError <- errors.Wrap(err, "sidechain")
case err := <-mainnnetErr:
intError <- errors.Wrap(err, "mainnet")
}
}()
// BUG: https://github.com/nspcc-dev/neofs-node/issues/346
s.mainnetListener.RegisterBlockHandler(func(b *block.Block) {
s.log.Debug("new block",
zap.Uint32("index", b.Index),
)
s.tickTimers()
})
go s.morphListener.ListenWithError(ctx, morphErr) // listen for neo:morph events
go s.mainnetListener.ListenWithError(ctx, mainnnetErr) // listen for neo:mainnet events
if err := s.startBlockTimers(); err != nil {
return errors.Wrap(err, "could not start block timers")
}
s.startWorkers(ctx)
2020-07-24 13:54:03 +00:00
return nil
}
func (s *Server) startWorkers(ctx context.Context) {
for _, w := range s.workers {
go w(ctx)
}
}
2020-07-24 13:54:03 +00:00
// Stop closes all subscription channels.
func (s *Server) Stop() {
go s.morphListener.Stop()
go s.mainnetListener.Stop()
}
// New creates instance of inner ring sever structure.
func New(ctx context.Context, log *zap.Logger, cfg *viper.Viper) (*Server, error) {
var err error
2020-07-24 13:54:03 +00:00
server := &Server{log: log}
// prepare inner ring node private key
server.key, err = crypto.LoadPrivateKey(cfg.GetString("key"))
2020-07-24 13:54:03 +00:00
if err != nil {
return nil, errors.Wrap(err, "ir: can't create private key")
}
// get all script hashes of contracts
server.contracts, err = parseContracts(cfg)
2020-07-24 13:54:03 +00:00
if err != nil {
return nil, err
}
// parse default validators
server.predefinedValidators, err = parsePredefinedValidators(cfg)
if err != nil {
return nil, errors.Wrap(err, "ir: can't parse predefined validators list")
}
2020-07-24 13:54:03 +00:00
morphChain := &chainParams{
log: log,
cfg: cfg,
key: server.key,
gas: server.contracts.gas,
2020-07-24 13:54:03 +00:00
name: morphPrefix,
}
// create morph listener
server.morphListener, err = createListener(ctx, morphChain)
if err != nil {
return nil, err
}
// create morph client
server.morphClient, err = createClient(ctx, morphChain)
if err != nil {
return nil, err
}
mainnetChain := morphChain
mainnetChain.name = mainnetPrefix
// create mainnet listener
server.mainnetListener, err = createListener(ctx, mainnetChain)
if err != nil {
return nil, err
}
// create mainnet client
server.mainnetClient, err = createClient(ctx, mainnetChain)
if err != nil {
return nil, err
}
server.pubKey = crypto.MarshalPublicKey(&server.key.PublicKey)
auditPool, err := ants.NewPool(cfg.GetInt("audit.task.exec_pool_size"))
if err != nil {
return nil, err
}
server.auditClient, err = invoke.NewNoFeeAuditClient(server.morphClient, server.contracts.audit)
if err != nil {
return nil, err
}
clientCache := newClientCache(&clientCacheParams{
Log: log,
Key: server.key,
SGTimeout: cfg.GetDuration("audit.timeout.get"),
HeadTimeout: cfg.GetDuration("audit.timeout.head"),
RangeTimeout: cfg.GetDuration("audit.timeout.rangehash"),
})
pdpPoolSize := cfg.GetInt("audit.pdp.pairs_pool_size")
porPoolSize := cfg.GetInt("audit.por.pool_size")
auditTaskManager := audittask.New(
audittask.WithQueueCapacity(cfg.GetUint32("audit.task.queue_capacity")),
audittask.WithWorkerPool(auditPool),
audittask.WithLogger(log),
audittask.WithContainerCommunicator(clientCache),
audittask.WithMaxPDPSleepInterval(cfg.GetDuration("audit.pdp.max_sleep_interval")),
audittask.WithPDPWorkerPoolGenerator(func() (util2.WorkerPool, error) {
return ants.NewPool(pdpPoolSize)
}),
audittask.WithPoRWorkerPoolGenerator(func() (util2.WorkerPool, error) {
return ants.NewPool(porPoolSize)
}),
)
server.workers = append(server.workers, auditTaskManager.Listen)
// create audit processor
auditProcessor, err := audit.New(&audit.Params{
Log: log,
NetmapContract: server.contracts.netmap,
ContainerContract: server.contracts.container,
AuditContract: server.contracts.audit,
MorphClient: server.morphClient,
IRList: server,
ClientCache: clientCache,
RPCSearchTimeout: cfg.GetDuration("audit.timeout.search"),
TaskManager: auditTaskManager,
Reporter: server,
})
if err != nil {
return nil, err
}
var netmapProcessor *netmap.Processor
server.epochTimer = timers.NewBlockTimer(
timers.StaticBlockMeter(cfg.GetUint32("timers.epoch")),
func() {
netmapProcessor.HandleNewEpochTick(timers.NewEpochTick{})
},
)
server.addBlockTimer(server.epochTimer)
2020-07-24 13:54:03 +00:00
// create netmap processor
netmapProcessor, err = netmap.New(&netmap.Params{
Log: log,
PoolSize: cfg.GetInt("workers.netmap"),
NetmapContract: server.contracts.netmap,
EpochTimer: (*blockTimerWrapper)(server.epochTimer),
MorphClient: server.morphClient,
EpochState: server,
ActiveState: server,
CleanupEnabled: cfg.GetBool("netmap_cleaner.enabled"),
CleanupThreshold: cfg.GetUint64("netmap_cleaner.threshold"),
HandleAudit: auditProcessor.StartAuditHandler(),
2020-07-24 13:54:03 +00:00
})
if err != nil {
return nil, err
}
err = bindMorphProcessor(netmapProcessor, server)
if err != nil {
return nil, err
}
// container processor
containerProcessor, err := container.New(&container.Params{
Log: log,
PoolSize: cfg.GetInt("workers.container"),
ContainerContract: server.contracts.container,
MorphClient: server.morphClient,
ActiveState: server,
})
if err != nil {
return nil, err
}
err = bindMorphProcessor(containerProcessor, server)
if err != nil {
return nil, err
}
2020-07-24 13:54:03 +00:00
// create balance processor
balanceProcessor, err := balance.New(&balance.Params{
Log: log,
PoolSize: cfg.GetInt("workers.balance"),
NeoFSContract: server.contracts.neofs,
BalanceContract: server.contracts.balance,
2020-07-24 13:54:03 +00:00
MainnetClient: server.mainnetClient,
ActiveState: server,
Converter: &server.precision,
2020-07-24 13:54:03 +00:00
})
if err != nil {
return nil, err
}
err = bindMorphProcessor(balanceProcessor, server)
if err != nil {
return nil, err
}
// todo: create reputation processor
// create mainnnet neofs processor
neofsProcessor, err := neofs.New(&neofs.Params{
Log: log,
PoolSize: cfg.GetInt("workers.neofs"),
NeoFSContract: server.contracts.neofs,
BalanceContract: server.contracts.balance,
NetmapContract: server.contracts.netmap,
MorphClient: server.morphClient,
EpochState: server,
ActiveState: server,
Converter: &server.precision,
MintEmitCacheSize: cfg.GetInt("emit.mint.cache_size"),
MintEmitThreshold: cfg.GetUint64("emit.mint.threshold"),
MintEmitValue: fixedn.Fixed8(cfg.GetInt64("emit.mint.value")),
2020-07-24 13:54:03 +00:00
})
if err != nil {
return nil, err
}
err = bindMainnetProcessor(neofsProcessor, server)
if err != nil {
return nil, err
}
var alphabetProcessor *alphabet.Processor
server.addBlockTimer(timers.NewBlockTimer(
timers.StaticBlockMeter(cfg.GetUint32("timers.emit")),
func() {
alphabetProcessor.HandleGasEmission(timers.NewAlphabetEmitTick{})
},
))
// create alphabet processor
alphabetProcessor, err = alphabet.New(&alphabet.Params{
Log: log,
PoolSize: cfg.GetInt("workers.alphabet"),
AlphabetContracts: server.contracts.alphabet,
NetmapContract: server.contracts.netmap,
MorphClient: server.morphClient,
IRList: server,
StorageEmission: cfg.GetUint64("emit.storage.amount"),
})
if err != nil {
return nil, err
}
err = bindMorphProcessor(alphabetProcessor, server)
if err != nil {
return nil, err
}
2020-07-24 13:54:03 +00:00
// todo: create vivid id component
return server, nil
}
func createListener(ctx context.Context, p *chainParams) (event.Listener, error) {
sub, err := subscriber.New(ctx, &subscriber.Params{
Log: p.log,
Endpoint: p.cfg.GetString(p.name + ".endpoint.notification"),
DialTimeout: p.cfg.GetDuration(p.name + ".dial_timeouts"),
})
if err != nil {
return nil, err
}
listener, err := event.NewListener(event.ListenerParams{
Logger: p.log,
Subscriber: sub,
})
if err != nil {
return nil, err
}
return listener, err
}
func createClient(ctx context.Context, p *chainParams) (*client.Client, error) {
return client.New(
p.key,
p.cfg.GetString(p.name+".endpoint.client"),
client.WithContext(ctx),
client.WithLogger(p.log),
client.WithDialTimeout(p.cfg.GetDuration(p.name+".dial_timeouts")),
client.WithGasContract(p.gas),
)
}
func parseContracts(cfg *viper.Viper) (*contracts, error) {
var (
result = new(contracts)
err error
)
netmapContractStr := cfg.GetString("contracts.netmap")
neofsContractStr := cfg.GetString("contracts.neofs")
balanceContractStr := cfg.GetString("contracts.balance")
nativeGasContractStr := cfg.GetString("contracts.gas")
containerContractStr := cfg.GetString("contracts.container")
auditContractStr := cfg.GetString("contracts.audit")
2020-07-24 13:54:03 +00:00
result.netmap, err = util.Uint160DecodeStringLE(netmapContractStr)
if err != nil {
return nil, errors.Wrap(err, "ir: can't read netmap script-hash")
}
result.neofs, err = util.Uint160DecodeStringLE(neofsContractStr)
if err != nil {
return nil, errors.Wrap(err, "ir: can't read neofs script-hash")
}
result.balance, err = util.Uint160DecodeStringLE(balanceContractStr)
if err != nil {
return nil, errors.Wrap(err, "ir: can't read balance script-hash")
}
result.gas, err = util.Uint160DecodeStringLE(nativeGasContractStr)
if err != nil {
return nil, errors.Wrap(err, "ir: can't read native gas script-hash")
}
result.container, err = util.Uint160DecodeStringLE(containerContractStr)
if err != nil {
return nil, errors.Wrap(err, "ir: can't read container script-hash")
}
result.audit, err = util.Uint160DecodeStringLE(auditContractStr)
if err != nil {
return nil, errors.Wrap(err, "ir: can't read audit script-hash")
}
result.alphabet, err = parseAlphabetContracts(cfg)
if err != nil {
return nil, err
}
2020-07-24 13:54:03 +00:00
return result, nil
}
func parsePredefinedValidators(cfg *viper.Viper) ([]keys.PublicKey, error) {
publicKeyStrings := cfg.GetStringSlice("morph.validators")
return ParsePublicKeysFromStrings(publicKeyStrings)
}
// ParsePublicKeysFromStrings returns slice of neo public keys from slice
// of hex encoded strings.
func ParsePublicKeysFromStrings(pubKeys []string) ([]keys.PublicKey, error) {
publicKeys := make([]keys.PublicKey, 0, len(pubKeys))
for i := range pubKeys {
key, err := keys.NewPublicKeyFromString(pubKeys[i])
if err != nil {
return nil, errors.Wrap(err, "can't decode public key")
}
publicKeys = append(publicKeys, *key)
}
return publicKeys, nil
}
func parseAlphabetContracts(cfg *viper.Viper) (res [7]util.Uint160, err error) {
// list of glagolic script letters that represent alphabet contracts
glagolic := []string{"az", "buky", "vedi", "glagoli", "dobro", "jest", "zhivete"}
for i, letter := range glagolic {
contractStr := cfg.GetString("contracts.alphabet." + letter)
res[i], err = util.Uint160DecodeStringLE(contractStr)
if err != nil {
return res, errors.Wrapf(err, "ir: can't read alphabet %s contract", letter)
}
}
return res, nil
}
func (s *Server) initConfigFromBlockchain() error {
2020-07-24 13:54:03 +00:00
// get current epoch
epoch, err := invoke.Epoch(s.morphClient, s.contracts.netmap)
2020-07-24 13:54:03 +00:00
if err != nil {
return errors.Wrap(err, "can't read epoch")
2020-07-24 13:54:03 +00:00
}
key := &s.key.PublicKey
// check if node inside inner ring list and what index it has
index, size, err := invoke.InnerRingIndex(s.mainnetClient, s.contracts.neofs, key)
2020-07-24 13:54:03 +00:00
if err != nil {
return errors.Wrap(err, "can't read inner ring list")
2020-07-24 13:54:03 +00:00
}
// get balance precision
balancePrecision, err := invoke.BalancePrecision(s.morphClient, s.contracts.balance)
if err != nil {
return errors.Wrap(err, "can't read balance contract precision")
}
2020-07-24 13:54:03 +00:00
s.epochCounter.Store(uint64(epoch))
s.innerRingSize.Store(size)
s.innerRingIndex.Store(index)
s.precision.SetBalancePrecision(balancePrecision)
2020-07-24 13:54:03 +00:00
s.log.Debug("read config from blockchain",
zap.Bool("active", s.IsActive()),
2020-07-24 13:54:03 +00:00
zap.Int64("epoch", epoch),
zap.Uint32("precision", balancePrecision),
2020-07-24 13:54:03 +00:00
)
return nil
}
func (s *Server) addBlockTimer(t *timers.BlockTimer) {
s.blockTimers = append(s.blockTimers, t)
}
func (s *Server) startBlockTimers() error {
for i := range s.blockTimers {
if err := s.blockTimers[i].Reset(); err != nil {
return err
}
}
return nil
}
func (s *Server) tickTimers() {
for i := range s.blockTimers {
s.blockTimers[i].Tick()
}
}