frostfs-node/pkg/services/container/announcement/load/controller/calls.go

package loadcontroller

import (
	"context"

	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
	"go.uber.org/zap"
)

// StartPrm groups the required parameters of the Controller.Start method.
type StartPrm struct {
	// Epoch number by which you want to select
	// the values of the used space of containers.
	Epoch uint64
}

type commonContext struct {
	epoch uint64

	ctrl *Controller

	log *logger.Logger
}

type announcer struct {
	commonContext
}

// Start starts the processing of container.SizeEstimation values.
//
// Single Start operation overtakes all data from LocalMetrics to
// LocalAnnouncementTarget (Controller's parameters).
// No filter by epoch is used for the iterator, since it is expected
// that the source of metrics does not track the change of epochs.
//
// Each call acquires an announcement context for an Epoch parameter.
// At the very end of the operation, the context is released.
func (c *Controller) Start(ctx context.Context, prm StartPrm) {
	var announcer *announcer
	// acquire announcement
	ctx, announcer = c.acquireAnnouncement(ctx, prm)
	if announcer == nil {
		return
	}

	// finally stop and free the announcement
	defer announcer.freeAnnouncement()

	// announce local values
	announcer.announce(ctx)
}

func (c *announcer) announce(ctx context.Context) {
	c.log.Debug("starting to announce the values of the metrics")

	var (
		metricsIterator Iterator
		err             error
	)

	// initialize iterator over locally collected metrics
	metricsIterator, err = c.ctrl.prm.LocalMetrics.InitIterator()
	if err != nil {
		c.log.Debug("could not initialize iterator over locally collected metrics",
			zap.String("error", err.Error()),
		)

		return
	}

	// initialize target of local announcements
	targetWriter, err := c.ctrl.prm.LocalAnnouncementTarget.InitWriter(nil)
	if err != nil {
		c.log.Debug("could not initialize announcement accumulator",
			zap.String("error", err.Error()),
		)

		return
	}

	// iterate over all collected metrics and write them to the target
	err = metricsIterator.Iterate(
		func(container.SizeEstimation) bool {
			return true // local metrics don't know about epochs
		},
		func(a container.SizeEstimation) error {
			a.SetEpoch(c.epoch) // set epoch explicitly
			return targetWriter.Put(a)
		},
	)
	if err != nil {
		c.log.Debug("iterator over locally collected metrics aborted",
			zap.String("error", err.Error()),
		)

		return
	}

	// finish writing
	err = targetWriter.Close(ctx)
	if err != nil {
		c.log.Debug("could not finish writing local announcements",
			zap.String("error", err.Error()),
		)

		return
	}

	c.log.Debug("trust announcement successfully finished")
}

func (c *Controller) acquireAnnouncement(ctx context.Context, prm StartPrm) (context.Context, *announcer) {
	started := true
	c.announceMtx.Lock()
	{
		if cancel := c.mAnnounceCtx[prm.Epoch]; cancel == nil {
			ctx, cancel = context.WithCancel(ctx)
			c.mAnnounceCtx[prm.Epoch] = cancel
			started = false
		}
	}
	c.announceMtx.Unlock()

	log := &logger.Logger{Logger: c.opts.log.With(
		zap.Uint64("epoch", prm.Epoch),
	)}

	if started {
		log.Debug("announcement is already started")
		return ctx, nil
	}

	return ctx, &announcer{
		commonContext: commonContext{
			epoch: prm.Epoch,
			ctrl:  c,
			log:   log,
		},
	}
}

func (c *commonContext) freeAnnouncement() {
	var stopped bool

	c.ctrl.announceMtx.Lock()

	{
		var cancel context.CancelFunc

		cancel, stopped = c.ctrl.mAnnounceCtx[c.epoch]

		if stopped {
			cancel()
			delete(c.ctrl.mAnnounceCtx, c.epoch)
		}
	}

	c.ctrl.announceMtx.Unlock()

	if stopped {
		c.log.Debug("announcement successfully interrupted")
	} else {
		c.log.Debug("announcement is not started or already interrupted")
	}
}

// StopPrm groups the required parameters of the Controller.Stop method.
type StopPrm struct {
	// Epoch number the analysis of the values of which must be interrupted.
	Epoch uint64
}

type reporter struct {
	commonContext
}

// Stop interrupts the processing of container.SizeEstimation values.
//
// Single Stop operation releases an announcement context and overtakes
// all data from AnnouncementAccumulator to ResultReceiver (Controller's
// parameters). Only values for the specified Epoch parameter are processed.
//
// Each call acquires a report context for an Epoch parameter.
// At the very end of the operation, the context is released.
func (c *Controller) Stop(ctx context.Context, prm StopPrm) {
	var reporter *reporter
	ctx, reporter = c.acquireReport(ctx, prm)
	if reporter == nil {
		return
	}

	// finally stop and free reporting
	defer reporter.freeReport()

	// interrupt announcement
	reporter.freeAnnouncement()

	// report the estimations
	reporter.report(ctx)
}

func (c *Controller) acquireReport(ctx context.Context, prm StopPrm) (context.Context, *reporter) {
	started := true

	c.reportMtx.Lock()
	{
		if cancel := c.mReportCtx[prm.Epoch]; cancel == nil {
			ctx, cancel = context.WithCancel(ctx)
			c.mReportCtx[prm.Epoch] = cancel
			started = false
		}
	}

	c.reportMtx.Unlock()

	log := &logger.Logger{Logger: c.opts.log.With(
		zap.Uint64("epoch", prm.Epoch),
	)}

	if started {
		log.Debug("report is already started")
		return ctx, nil
	}

	return ctx, &reporter{
		commonContext: commonContext{
			epoch: prm.Epoch,
			ctrl:  c,
			log:   log,
		},
	}
}

func (c *commonContext) freeReport() {
	var stopped bool

	c.ctrl.reportMtx.Lock()

	{
		var cancel context.CancelFunc

		cancel, stopped = c.ctrl.mReportCtx[c.epoch]

		if stopped {
			cancel()
			delete(c.ctrl.mReportCtx, c.epoch)
		}
	}

	c.ctrl.reportMtx.Unlock()

	if stopped {
		c.log.Debug("announcement successfully interrupted")
	} else {
		c.log.Debug("announcement is not started or already interrupted")
	}
}

func (c *reporter) report(ctx context.Context) {
	var (
		localIterator Iterator
		err           error
	)

	// initialize iterator over locally accumulated announcements
	localIterator, err = c.ctrl.prm.AnnouncementAccumulator.InitIterator()
	if err != nil {
		c.log.Debug("could not initialize iterator over locally accumulated announcements",
			zap.String("error", err.Error()),
		)

		return
	}

	// initialize final destination of load estimations
	resultWriter, err := c.ctrl.prm.ResultReceiver.InitWriter(nil)
	if err != nil {
		c.log.Debug("could not initialize result target",
			zap.String("error", err.Error()),
		)

		return
	}

	// iterate over all accumulated announcements and write them to the target
	err = localIterator.Iterate(
		usedSpaceFilterEpochEQ(c.epoch),
		resultWriter.Put,
	)
	if err != nil {
		c.log.Debug("iterator over local announcements aborted",
			zap.String("error", err.Error()),
		)

		return
	}

	// finish writing
	err = resultWriter.Close(ctx)
	if err != nil {
		c.log.Debug("could not finish writing load estimations",
			zap.String("error", err.Error()),
		)
	}
}