From 4dc09b19f3f2ffbb5b21ac43d1d6d2bc8208a0cd Mon Sep 17 00:00:00 2001 From: Leonard Lyubich Date: Tue, 22 Dec 2020 03:28:42 +0300 Subject: [PATCH] [#255] ir: Make audit processor to push tasks to audit task manager Signed-off-by: Leonard Lyubich --- pkg/innerring/innerring.go | 48 +++++++++++++++++++++ pkg/innerring/processors/audit/process.go | 23 +++++++++- pkg/innerring/processors/audit/processor.go | 29 +++++++++++++ pkg/innerring/processors/audit/scheduler.go | 4 ++ 4 files changed, 103 insertions(+), 1 deletion(-) diff --git a/pkg/innerring/innerring.go b/pkg/innerring/innerring.go index 703052195..9902821b0 100644 --- a/pkg/innerring/innerring.go +++ b/pkg/innerring/innerring.go @@ -17,9 +17,14 @@ import ( "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" + auditClient "github.com/nspcc-dev/neofs-node/pkg/morph/client/audit" + auditWrapper "github.com/nspcc-dev/neofs-node/pkg/morph/client/audit/wrapper" "github.com/nspcc-dev/neofs-node/pkg/morph/event" "github.com/nspcc-dev/neofs-node/pkg/morph/subscriber" + auditSvc "github.com/nspcc-dev/neofs-node/pkg/services/audit" + audittask "github.com/nspcc-dev/neofs-node/pkg/services/audit/taskmanager" "github.com/nspcc-dev/neofs-node/pkg/util/precision" + "github.com/panjf2000/ants/v2" "github.com/pkg/errors" "github.com/spf13/viper" "go.uber.org/atomic" @@ -44,11 +49,15 @@ type ( innerRingIndex atomic.Int32 innerRingSize atomic.Int32 precision precision.Fixed8Converter + auditClient *auditWrapper.ClientWrapper // internal variables key *ecdsa.PrivateKey + pubKey []byte contracts *contracts predefinedValidators []keys.PublicKey + + workers []func(context.Context) } contracts struct { @@ -115,15 +124,30 @@ func (s *Server) Start(ctx context.Context, intError chan<- error) error { go s.morphListener.ListenWithError(ctx, morphErr) // listen for neo:morph events go s.mainnetListener.ListenWithError(ctx, mainnnetErr) // listen for neo:mainnet events + s.startWorkers(ctx) + return nil } +func (s *Server) startWorkers(ctx context.Context) { + for _, w := range s.workers { + go w(ctx) + } +} + // Stop closes all subscription channels. func (s *Server) Stop() { go s.morphListener.Stop() go s.mainnetListener.Stop() } +func (s *Server) WriteReport(r *auditSvc.Report) error { + res := r.Result() + res.SetPublicKey(s.pubKey) + + return s.auditClient.PutAuditResult(res) +} + // New creates instance of inner ring sever structure. func New(ctx context.Context, log *zap.Logger, cfg *viper.Viper) (*Server, error) { var err error @@ -189,6 +213,28 @@ func New(ctx context.Context, log *zap.Logger, cfg *viper.Viper) (*Server, error return nil, err } + server.pubKey = crypto.MarshalPublicKey(&server.key.PublicKey) + + auditPool, err := ants.NewPool(cfg.GetInt("audit.task.exec_pool_size"), ants.WithNonblocking(true)) + if err != nil { + return nil, err + } + + staticAuditClient, err := client.NewStatic(server.morphClient, server.contracts.audit, 0) + if err != nil { + return nil, err + } + + server.auditClient = auditWrapper.WrapClient(auditClient.New(staticAuditClient)) + + auditTaskManager := audittask.New( + audittask.WithQueueCapacity(cfg.GetUint32("audit.task.queue_capacity")), + audittask.WithWorkerPool(auditPool), + audittask.WithLogger(log), + ) + + server.workers = append(server.workers, auditTaskManager.Listen) + // create audit processor auditProcessor, err := audit.New(&audit.Params{ Log: log, @@ -198,6 +244,8 @@ func New(ctx context.Context, log *zap.Logger, cfg *viper.Viper) (*Server, error MorphClient: server.morphClient, IRList: server, ClientCache: newClientCache(server.key), + TaskManager: auditTaskManager, + Reporter: server, }) if err != nil { return nil, err diff --git a/pkg/innerring/processors/audit/process.go b/pkg/innerring/processors/audit/process.go index 36b2012cb..82b48e3d2 100644 --- a/pkg/innerring/processors/audit/process.go +++ b/pkg/innerring/processors/audit/process.go @@ -10,6 +10,7 @@ import ( "github.com/nspcc-dev/neofs-api-go/pkg/netmap" "github.com/nspcc-dev/neofs-api-go/pkg/object" "github.com/nspcc-dev/neofs-node/pkg/network" + "github.com/nspcc-dev/neofs-node/pkg/services/audit" "github.com/nspcc-dev/neofs-node/pkg/services/object_manager/storagegroup" "go.uber.org/zap" ) @@ -19,6 +20,8 @@ var sgFilter = storagegroup.SearchQuery() func (ap *Processor) processStartAudit(epoch uint64) { log := ap.log.With(zap.Uint64("epoch", epoch)) + ap.prevAuditCanceler() + containers, err := ap.selectContainersToAudit(epoch) if err != nil { log.Error("container selection failure", zap.String("error", err.Error())) @@ -68,7 +71,25 @@ func (ap *Processor) processStartAudit(epoch uint64) { zap.Stringer("cid", containers[i]), zap.Int("amount", len(storageGroups))) - // todo: for each container push audit tasks into queue + var auditCtx context.Context + auditCtx, ap.prevAuditCanceler = context.WithCancel(context.Background()) + + auditTask := new(audit.Task). + WithReporter(&epochAuditReporter{ + epoch: epoch, + rep: ap.reporter, + }). + WithAuditContext(auditCtx). + WithContainerID(containers[i]). + WithStorageGroupList(storageGroups). + WithContainerStructure(cnr). + WithContainerNodes(nodes) + + if err := ap.taskManager.PushTask(auditTask); err != nil { + ap.log.Error("could not push audit task", + zap.String("error", err.Error()), + ) + } } } diff --git a/pkg/innerring/processors/audit/processor.go b/pkg/innerring/processors/audit/processor.go index 949402331..e38adb5a7 100644 --- a/pkg/innerring/processors/audit/processor.go +++ b/pkg/innerring/processors/audit/processor.go @@ -1,6 +1,8 @@ package audit import ( + "context" + "github.com/nspcc-dev/neo-go/pkg/util" SDKClient "github.com/nspcc-dev/neofs-api-go/pkg/client" "github.com/nspcc-dev/neofs-node/pkg/innerring/invoke" @@ -8,6 +10,7 @@ import ( wrapContainer "github.com/nspcc-dev/neofs-node/pkg/morph/client/container/wrapper" wrapNetmap "github.com/nspcc-dev/neofs-node/pkg/morph/client/netmap/wrapper" "github.com/nspcc-dev/neofs-node/pkg/morph/event" + "github.com/nspcc-dev/neofs-node/pkg/services/audit" "github.com/panjf2000/ants/v2" "github.com/pkg/errors" "go.uber.org/zap" @@ -25,6 +28,10 @@ type ( Get(address string, opts ...SDKClient.Option) (*SDKClient.Client, error) } + TaskManager interface { + PushTask(*audit.Task) error + } + // Processor of events related with data audit. Processor struct { log *zap.Logger @@ -37,6 +44,10 @@ type ( containerClient *wrapContainer.Wrapper netmapClient *wrapNetmap.Wrapper + + taskManager TaskManager + reporter audit.Reporter + prevAuditCanceler context.CancelFunc } // Params of the processor constructor. @@ -48,9 +59,17 @@ type ( MorphClient *client.Client IRList Indexer ClientCache NeoFSClientCache + TaskManager TaskManager + Reporter audit.Reporter } ) +type epochAuditReporter struct { + epoch uint64 + + rep audit.Reporter +} + // AuditProcessor manages audit tasks and fills queue for next epoch. This // process must not be interrupted by new audit epoch, so we limit pool size // for processor to one. @@ -96,6 +115,9 @@ func New(p *Params) (*Processor, error) { clientCache: p.ClientCache, containerClient: containerClient, netmapClient: netmapClient, + taskManager: p.TaskManager, + reporter: p.Reporter, + prevAuditCanceler: func() {}, }, nil } @@ -118,3 +140,10 @@ func (ap *Processor) TimersHandlers() []event.HandlerInfo { func (ap *Processor) StartAuditHandler() event.Handler { return ap.handleNewAuditRound } + +func (r *epochAuditReporter) WriteReport(rep *audit.Report) error { + res := rep.Result() + res.SetAuditEpoch(r.epoch) + + return r.rep.WriteReport(rep) +} diff --git a/pkg/innerring/processors/audit/scheduler.go b/pkg/innerring/processors/audit/scheduler.go index 701308bf2..c4aa711c4 100644 --- a/pkg/innerring/processors/audit/scheduler.go +++ b/pkg/innerring/processors/audit/scheduler.go @@ -6,6 +6,7 @@ import ( "github.com/nspcc-dev/neofs-api-go/pkg/container" "github.com/pkg/errors" + "go.uber.org/zap" ) var ErrInvalidIRNode = errors.New("node is not in the inner ring list") @@ -18,6 +19,9 @@ func (ap *Processor) selectContainersToAudit(epoch uint64) ([]*container.ID, err // consider getting extra information about container complexity from // audit contract there + ap.log.Debug("container listing finished", + zap.Int("total amount", len(containers)), + ) sort.Slice(containers, func(i, j int) bool { return strings.Compare(containers[i].String(), containers[j].String()) < 0