98 lines
3.5 KiB
Go
98 lines
3.5 KiB
Go
package handlers
|
|
|
|
import (
|
|
"github.com/fiskerinc/cloud-services/services/jetfire/models"
|
|
"github.com/fiskerinc/cloud-services/services/jetfire/services"
|
|
"github.com/fiskerinc/cloud-services/services/jetfire/utils"
|
|
"time"
|
|
|
|
"github.com/fiskerinc/cloud-services/pkg/grpc/kafka_grpc"
|
|
"github.com/fiskerinc/cloud-services/pkg/logger"
|
|
"github.com/fiskerinc/cloud-services/pkg/utils/envtool"
|
|
)
|
|
|
|
var (
|
|
//downsampling timers and delays
|
|
clearCacheTimer = time.Now().Truncate(time.Second)
|
|
featureDelay = envtool.GetEnvDuration("JETFIRE_FEATURE_DOWNSAMPLE_US", 1000000) * time.Microsecond
|
|
clearCacheDelay = envtool.GetEnvDuration("JETFIRE_STATE_TIMEOUT_MS", 3600000) * time.Millisecond
|
|
futureTimeThreshold = envtool.GetEnvDuration("JETFIRE_FUTURE_THRESHOLD_MS", 2*24*60*60*1000) * time.Millisecond
|
|
)
|
|
|
|
// Handles the batch of Signals. returns batchFlag, error
|
|
// where batchFlag corresponds to FeatureUpdateFlag for which buffers were updated with this batch
|
|
func HandleSignalBatch(batchData []*kafka_grpc.GRPC_CANSignal, vehicleCache *models.VehicleCache, producerChannel chan models.InsertCommand) (uint, error) {
|
|
start := time.Now()
|
|
|
|
batchFlag := uint(0)
|
|
|
|
//Iterate through received can signals and add to cache
|
|
futureWarning := false //only one future warning per batch of data
|
|
|
|
//sets of pointers to insert into insertion buffers
|
|
featureUpdates := make(map[*models.VehicleState]bool)
|
|
|
|
logger.Debug().Msgf("Processing batch %d signals...", len(batchData))
|
|
|
|
skipLog := false
|
|
|
|
for _, signal := range batchData {
|
|
signal.Timestamp = utils.FixFloatTimestampScale(signal.Timestamp)
|
|
if start.Add(futureTimeThreshold).Before(utils.FloatToTime(signal.Timestamp)) && !futureWarning {
|
|
// Throw out any signals that are from the future.
|
|
futureWarning = true
|
|
logger.Warn().Msgf("ignoring signal(s) from %s from future: %f (currently %f)", signal.Vin, signal.Timestamp, utils.TimeToFloat(start))
|
|
continue
|
|
}
|
|
|
|
err := vehicleCache.UpdateSignal(signal, utils.FeatureUpdateFlag)
|
|
if err != nil {
|
|
logger.Error().Err(err).Send()
|
|
//do not continue yet; attempt to add to vehicle signal buffer
|
|
}
|
|
|
|
//Check vehicle state for update
|
|
// if signal is not in tracked vars, then skip.
|
|
// if VIN is not in vehicle cache, then log and skip
|
|
state, ok := vehicleCache.Cache[signal.Vin]
|
|
if !ok || state == nil {
|
|
if len(*vehicleCache.SignalsSet) > 0 {
|
|
_, ok = (*vehicleCache.SignalsSet)[signal.Name]
|
|
if ok && !skipLog {
|
|
// no vin state but signal update should have created a vin state... log warning
|
|
logger.Error().Msgf("unexpected missing VIN state from cache for VIN {%s} and signal {%s}", signal.Vin, signal.Name)
|
|
skipLog = true
|
|
}
|
|
}
|
|
continue
|
|
}
|
|
if state.TimeSincePolled(utils.FeatureUpdateFlag) > featureDelay {
|
|
featureUpdates[state] = true
|
|
}
|
|
}
|
|
|
|
services.SchemaLock.Lock()
|
|
defer services.SchemaLock.Unlock()
|
|
|
|
//TODO: investigate optimizing selecting VINs that need to be appended to buffer
|
|
// iterating through maps is slow!
|
|
|
|
//batch rows for insertion for feature
|
|
for vinState := range featureUpdates {
|
|
err := services.GetFeatureBatch().AppendRow(services.GetFeatureVars(), vinState, producerChannel)
|
|
if err != nil {
|
|
logger.Error().Err(err).Send()
|
|
} else {
|
|
vinState.SetPollTime(utils.FeatureUpdateFlag)
|
|
batchFlag |= utils.FeatureUpdateFlag
|
|
}
|
|
|
|
err = services.GetFeatureLastBatch().AppendRow(services.GetFeatureVars(), vinState, producerChannel)
|
|
if err != nil {
|
|
logger.Error().Err(err).Send()
|
|
}
|
|
}
|
|
|
|
return batchFlag, nil
|
|
}
|