frostfs-node/pkg/innerring/processors/frostfs/handlers.go

package frostfs

import (
	"encoding/hex"

	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/innerring/processors"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event"
	frostfsEvent "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event/frostfs"
	"github.com/nspcc-dev/neo-go/pkg/util/slice"
	"go.uber.org/zap"
)

func (np *Processor) handleDeposit(ev event.Event) {
	deposit := ev.(frostfsEvent.Deposit)
	np.log.Info(logs.Notification,
		zap.String("type", "deposit"),
		zap.String("id", hex.EncodeToString(slice.CopyReverse(deposit.ID()))))

	// send event to the worker pool

	err := processors.SubmitEvent(np.pool, np.metrics, "frostfs_deposit", func() bool {
		return np.processDeposit(deposit)
	})
	if err != nil {
		// there system can be moved into controlled degradation stage
		np.log.Warn(logs.FrostFSFrostfsProcessorWorkerPoolDrained,
			zap.Int("capacity", np.pool.Cap()))
	}
}

func (np *Processor) handleWithdraw(ev event.Event) {
	withdraw := ev.(frostfsEvent.Withdraw)
	np.log.Info(logs.Notification,
		zap.String("type", "withdraw"),
		zap.String("id", hex.EncodeToString(slice.CopyReverse(withdraw.ID()))))

	// send event to the worker pool

	err := processors.SubmitEvent(np.pool, np.metrics, "frostfs_withdraw", func() bool {
		return np.processWithdraw(withdraw)
	})
	if err != nil {
		// there system can be moved into controlled degradation stage
		np.log.Warn(logs.FrostFSFrostfsProcessorWorkerPoolDrained,
			zap.Int("capacity", np.pool.Cap()))
	}
}

func (np *Processor) handleCheque(ev event.Event) {
	cheque := ev.(frostfsEvent.Cheque)
	np.log.Info(logs.Notification,
		zap.String("type", "cheque"),
		zap.String("id", hex.EncodeToString(cheque.ID())))

	// send event to the worker pool

	err := processors.SubmitEvent(np.pool, np.metrics, "frostfs_cheque", func() bool {
		return np.processCheque(cheque)
	})
	if err != nil {
		// there system can be moved into controlled degradation stage
		np.log.Warn(logs.FrostFSFrostfsProcessorWorkerPoolDrained,
			zap.Int("capacity", np.pool.Cap()))
	}
}

func (np *Processor) handleConfig(ev event.Event) {
	cfg := ev.(frostfsEvent.Config)
	np.log.Info(logs.Notification,
		zap.String("type", "set config"),
		zap.String("key", hex.EncodeToString(cfg.Key())),
		zap.String("value", hex.EncodeToString(cfg.Value())))

	// send event to the worker pool

	err := processors.SubmitEvent(np.pool, np.metrics, "frostfs_config", func() bool {
		return np.processConfig(cfg)
	})
	if err != nil {
		// there system can be moved into controlled degradation stage
		np.log.Warn(logs.FrostFSFrostfsProcessorWorkerPoolDrained,
			zap.Int("capacity", np.pool.Cap()))
	}
}

func (np *Processor) handleBind(ev event.Event) {
	e := ev.(frostfsEvent.Bind)
	np.log.Info(logs.Notification,
		zap.String("type", "bind"),
	)

	// send event to the worker pool

	err := processors.SubmitEvent(np.pool, np.metrics, "frostfs_bind", func() bool {
		return np.processBind(e, true)
	})
	if err != nil {
		// there system can be moved into controlled degradation stage
		np.log.Warn(logs.FrostFSFrostfsProcessorWorkerPoolDrained,
			zap.Int("capacity", np.pool.Cap()))
	}
}

func (np *Processor) handleUnbind(ev event.Event) {
	e := ev.(frostfsEvent.Unbind)
	np.log.Info(logs.Notification,
		zap.String("type", "unbind"),
	)

	// send event to the worker pool

	err := processors.SubmitEvent(np.pool, np.metrics, "frostfs_unbind", func() bool {
		return np.processBind(e, false)
	})
	if err != nil {
		// there system can be moved into controlled degradation stage
		np.log.Warn(logs.FrostFSFrostfsProcessorWorkerPoolDrained,
			zap.Int("capacity", np.pool.Cap()))
	}
}