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

package container

import (
	"context"
	"crypto/sha256"

	"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"
	containerEvent "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event/container"
	"github.com/mr-tron/base58"
	"go.uber.org/zap"
)

func (cp *Processor) handlePut(ctx context.Context, ev event.Event) {
	put := ev.(putEvent)

	id := sha256.Sum256(put.Container())
	cp.log.Info(ctx, logs.Notification,
		zap.String("type", "container put"),
		zap.String("id", base58.Encode(id[:])))

	// send an event to the worker pool

	err := processors.SubmitEvent(cp.pool, cp.metrics, "container_put", func() bool {
		return cp.processContainerPut(ctx, put)
	})
	if err != nil {
		// there system can be moved into controlled degradation stage
		cp.log.Warn(ctx, logs.ContainerContainerProcessorWorkerPoolDrained,
			zap.Int("capacity", cp.pool.Cap()))
	}
}

func (cp *Processor) handleDelete(ctx context.Context, ev event.Event) {
	del := ev.(containerEvent.Delete)
	cp.log.Info(ctx, logs.Notification,
		zap.String("type", "container delete"),
		zap.String("id", base58.Encode(del.ContainerID())))

	// send an event to the worker pool

	err := processors.SubmitEvent(cp.pool, cp.metrics, "container_delete", func() bool {
		return cp.processContainerDelete(ctx, del)
	})
	if err != nil {
		// there system can be moved into controlled degradation stage
		cp.log.Warn(ctx, logs.ContainerContainerProcessorWorkerPoolDrained,
			zap.Int("capacity", cp.pool.Cap()))
	}
}