91 lines
2.7 KiB
Go
91 lines
2.7 KiB
Go
package frostfs
|
|
|
|
import (
|
|
"bytes"
|
|
"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"
|
|
"go.uber.org/zap"
|
|
"golang.org/x/exp/slices"
|
|
)
|
|
|
|
func (np *Processor) handleDeposit(ev event.Event) {
|
|
deposit := ev.(frostfsEvent.Deposit)
|
|
depositIDBin := bytes.Clone(deposit.ID())
|
|
slices.Reverse(depositIDBin)
|
|
np.log.Info(logs.Notification,
|
|
zap.String("type", "deposit"),
|
|
zap.String("id", hex.EncodeToString(depositIDBin)))
|
|
|
|
// 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)
|
|
withdrawBin := bytes.Clone(withdraw.ID())
|
|
slices.Reverse(withdrawBin)
|
|
np.log.Info(logs.Notification,
|
|
zap.String("type", "withdraw"),
|
|
zap.String("id", hex.EncodeToString(withdrawBin)))
|
|
|
|
// 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()))
|
|
}
|
|
}
|