[#255] ir: Make audit processor to push tasks to audit task manager

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>

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

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()),
+			)
+		}
 	}
 }
 

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)
+}

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