modelRT/pool/concurrency_anchor_parse.go

// Package pool define concurrency call function in ants
package pool

import (
	"fmt"
	"time"

	"modelRT/alert"
	"modelRT/config"
	"modelRT/constant"
	"modelRT/diagram"
	"modelRT/logger"

	"github.com/panjf2000/ants/v2"
	"go.uber.org/zap"
)

// AnchorRealTimePool define anchor param pool of real time data
var AnchorRealTimePool *ants.PoolWithFunc

// AnchorPoolInit define anchor param pool init
func AnchorPoolInit(concurrentQuantity int) (pool *ants.PoolWithFunc, err error) {
	// init anchor param ants pool
	AnchorRealTimePool, err = ants.NewPoolWithFunc(concurrentQuantity, AnchorFunc)
	if err != nil {
		return nil, err
	}
	return AnchorRealTimePool, nil
}

// AnchorFunc defines func that process the real time data of component anchor params
var AnchorFunc = func(poolConfig interface{}) {
	var firstStart bool
	logger := logger.GetLoggerInstance()
	alertManager := alert.GetAlertMangerInstance()

	anchorChanConfig, ok := poolConfig.(config.AnchorChanConfig)
	if !ok {
		logger.Error("conversion component anchor chan type failed")
		return
	}

	for {
		select {
		case <-anchorChanConfig.Ctx.Done():
			return
		case anchorParaConfig := <-anchorChanConfig.AnchorChan:
			if firstStart {
				close(anchorChanConfig.ReadyChan)
				firstStart = false
			}

			componentID := anchorParaConfig.ComponentID
			anchorRealTimeDatas := anchorParaConfig.AnchorRealTimeData

			for _, value := range anchorRealTimeDatas {
				anchorName, err := diagram.GetAnchorValue(componentID)
				if err != nil {
					logger.Error("can not get anchor value from map by uuid", zap.Int64("component_id", componentID), zap.Error(err))
					continue
				}

				if anchorName != anchorParaConfig.AnchorName {
					logger.Error("anchor name not equal param config anchor value", zap.String("map_anchor_name", anchorName), zap.String("param_anchor_name", anchorParaConfig.AnchorName))
					continue
				}

				upperLimitVal := anchorParaConfig.CompareValUpperLimit
				lowerlimitVal := anchorParaConfig.CompareValLowerLimit
				compareValue := anchorParaConfig.CalculateFunc(value, anchorParaConfig.CalculateParams...)
				if compareValue > upperLimitVal || compareValue < lowerlimitVal {
					message := fmt.Sprintf("anchor param %s value out of range, value:%f, upper limit:%f, lower limit:%f", anchorName,
						compareValue, upperLimitVal, lowerlimitVal)
					event := alert.Event{
						ComponentID: componentID,
						AnchorName:  anchorName,
						Level:       constant.InfoAlertLevel,
						Message:     message,
						StartTime:   time.Now().Unix(),
					}
					alertManager.AddEvent(event)
				}
			}
		default:
		}
	}
}