272 lines
10 KiB
Go
272 lines
10 KiB
Go
package platform
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
const alertStreamMaxFutureSkew = 10 * time.Minute
|
|
|
|
type AlertStreamRecord struct {
|
|
Topic string
|
|
Partition int
|
|
Offset int64
|
|
HighWatermark int64
|
|
Protocol string
|
|
VIN string
|
|
Plate string
|
|
SourceEventID string
|
|
EventAt time.Time
|
|
ReceivedAt time.Time
|
|
Fields map[string]json.RawMessage
|
|
Valid bool
|
|
Late bool
|
|
ErrorCode string
|
|
}
|
|
|
|
type AlertStreamBatchResult struct {
|
|
Fetched int
|
|
Processed int
|
|
Valid int
|
|
Invalid int
|
|
Late int
|
|
ReplaySkipped int
|
|
Partitions int
|
|
RulesEvaluated int
|
|
CandidatesAdvanced int
|
|
DuplicateObservations int
|
|
LateObservations int
|
|
Opened int
|
|
Recovered int
|
|
}
|
|
|
|
type alertStreamEnvelope struct {
|
|
EventID string `json:"event_id"`
|
|
EventKind string `json:"event_kind"`
|
|
SourceEventID string `json:"source_event_id"`
|
|
FieldMapping string `json:"field_mapping"`
|
|
Protocol string `json:"protocol"`
|
|
VIN string `json:"vin"`
|
|
Plate string `json:"plate"`
|
|
EventTimeMS int64 `json:"event_time_ms"`
|
|
ReceivedAtMS int64 `json:"received_at_ms"`
|
|
Fields map[string]json.RawMessage `json:"fields"`
|
|
}
|
|
|
|
func DecodeAlertStreamRecord(topic string, partition int, offset, highWatermark int64, payload []byte, lateness time.Duration) AlertStreamRecord {
|
|
record := AlertStreamRecord{Topic: strings.TrimSpace(topic), Partition: partition, Offset: offset, HighWatermark: highWatermark}
|
|
var envelope alertStreamEnvelope
|
|
if err := json.Unmarshal(payload, &envelope); err != nil {
|
|
record.ErrorCode = "invalid_json"
|
|
return record
|
|
}
|
|
record.Protocol = strings.TrimSpace(envelope.Protocol)
|
|
record.VIN = strings.TrimSpace(envelope.VIN)
|
|
record.Plate = strings.TrimSpace(envelope.Plate)
|
|
record.SourceEventID = firstNonEmpty(strings.TrimSpace(envelope.SourceEventID), strings.TrimSpace(envelope.EventID))
|
|
record.Fields = envelope.Fields
|
|
if expected, known := alertStreamTopicProtocol(record.Topic); known && !strings.EqualFold(expected, record.Protocol) {
|
|
record.ErrorCode = "protocol_topic_mismatch"
|
|
return record
|
|
}
|
|
if !strings.EqualFold(strings.TrimSpace(envelope.EventKind), "FIELDS") {
|
|
record.ErrorCode = "event_kind_mismatch"
|
|
return record
|
|
}
|
|
if strings.TrimSpace(envelope.FieldMapping) == "" {
|
|
record.ErrorCode = "missing_field_mapping"
|
|
return record
|
|
}
|
|
if record.VIN == "" {
|
|
record.ErrorCode = "missing_vin_" + strings.ToLower(record.Protocol)
|
|
return record
|
|
}
|
|
if record.SourceEventID == "" {
|
|
record.ErrorCode = "missing_source_event_id"
|
|
return record
|
|
}
|
|
if len(envelope.Fields) == 0 {
|
|
record.ErrorCode = "missing_fields"
|
|
return record
|
|
}
|
|
prefix := map[string]string{"GB32960": "gb32960.", "JT808": "jt808.", "YUTONG_MQTT": "yutong_mqtt."}[strings.ToUpper(record.Protocol)]
|
|
for field := range envelope.Fields {
|
|
if prefix != "" && (field != strings.TrimSpace(field) || !strings.HasPrefix(field, prefix) || len(field) <= len(prefix)) {
|
|
record.ErrorCode = "invalid_field_name"
|
|
return record
|
|
}
|
|
}
|
|
if envelope.ReceivedAtMS <= 0 {
|
|
record.ErrorCode = "missing_received_time"
|
|
return record
|
|
}
|
|
receivedAt := time.UnixMilli(envelope.ReceivedAtMS)
|
|
eventMS := envelope.EventTimeMS
|
|
if eventMS <= 0 || eventMS > envelope.ReceivedAtMS+alertStreamMaxFutureSkew.Milliseconds() {
|
|
eventMS = envelope.ReceivedAtMS
|
|
}
|
|
record.EventAt = time.UnixMilli(eventMS)
|
|
record.ReceivedAt = receivedAt
|
|
if lateness > 0 && receivedAt.Sub(record.EventAt) > lateness {
|
|
record.Late = true
|
|
}
|
|
record.Valid = true
|
|
return record
|
|
}
|
|
|
|
func alertStreamTopicProtocol(topic string) (string, bool) {
|
|
switch strings.TrimSpace(topic) {
|
|
case "vehicle.fields.go.gb32960.v1":
|
|
return "GB32960", true
|
|
case "vehicle.fields.go.jt808.v1":
|
|
return "JT808", true
|
|
case "vehicle.fields.go.yutong-mqtt.v1":
|
|
return "YUTONG_MQTT", true
|
|
default:
|
|
return "", false
|
|
}
|
|
}
|
|
|
|
type alertStreamPartition struct {
|
|
topic string
|
|
partition int
|
|
}
|
|
|
|
type alertStreamCheckpointUpdate struct {
|
|
key alertStreamPartition
|
|
nextOffset, highWatermark int64
|
|
processed, valid, invalid, late, replay int64
|
|
lastEventAt, lastReceivedAt any
|
|
lastSourceEventID string
|
|
lastInvalidCode string
|
|
}
|
|
|
|
func (s *ProductionStore) RecordAlertStreamBatch(ctx context.Context, consumerGroup string, records []AlertStreamRecord) (AlertStreamBatchResult, error) {
|
|
result := AlertStreamBatchResult{Fetched: len(records)}
|
|
consumerGroup = strings.TrimSpace(consumerGroup)
|
|
if consumerGroup == "" || len(consumerGroup) > 128 {
|
|
return result, fmt.Errorf("alert stream consumer group is empty or too long")
|
|
}
|
|
if len(records) == 0 {
|
|
return result, nil
|
|
}
|
|
grouped := map[alertStreamPartition][]AlertStreamRecord{}
|
|
for _, record := range records {
|
|
if record.Topic == "" || len(record.Topic) > 255 || record.Partition < 0 || record.Offset < 0 {
|
|
return result, fmt.Errorf("invalid alert stream message position topic=%q partition=%d offset=%d", record.Topic, record.Partition, record.Offset)
|
|
}
|
|
key := alertStreamPartition{topic: record.Topic, partition: record.Partition}
|
|
grouped[key] = append(grouped[key], record)
|
|
}
|
|
keys := make([]alertStreamPartition, 0, len(grouped))
|
|
for key := range grouped {
|
|
keys = append(keys, key)
|
|
}
|
|
sort.Slice(keys, func(i, j int) bool {
|
|
if keys[i].topic == keys[j].topic {
|
|
return keys[i].partition < keys[j].partition
|
|
}
|
|
return keys[i].topic < keys[j].topic
|
|
})
|
|
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
defer tx.Rollback()
|
|
updates := make([]alertStreamCheckpointUpdate, 0, len(keys))
|
|
accepted := make([]AlertStreamRecord, 0, len(records))
|
|
for _, key := range keys {
|
|
partitionRecords := grouped[key]
|
|
sort.Slice(partitionRecords, func(i, j int) bool { return partitionRecords[i].Offset < partitionRecords[j].Offset })
|
|
var nextOffset int64
|
|
hasCheckpoint := true
|
|
if err = tx.QueryRowContext(ctx, `SELECT next_offset FROM vehicle_alert_stream_checkpoint WHERE consumer_group=? AND topic=? AND partition_id=? FOR UPDATE`, consumerGroup, key.topic, key.partition).Scan(&nextOffset); err == sql.ErrNoRows {
|
|
hasCheckpoint = false
|
|
nextOffset = -1
|
|
} else if err != nil {
|
|
return result, err
|
|
}
|
|
update := alertStreamCheckpointUpdate{key: key}
|
|
maxOffset := nextOffset - 1
|
|
var lastEventTime time.Time
|
|
var lastPosition int64
|
|
hasPosition := false
|
|
for _, record := range partitionRecords {
|
|
if record.HighWatermark > update.highWatermark {
|
|
update.highWatermark = record.HighWatermark
|
|
}
|
|
if hasCheckpoint && record.Offset < nextOffset {
|
|
update.replay++
|
|
continue
|
|
}
|
|
if hasPosition && record.Offset == lastPosition {
|
|
update.replay++
|
|
continue
|
|
}
|
|
lastPosition = record.Offset
|
|
hasPosition = true
|
|
update.processed++
|
|
if record.Valid {
|
|
update.valid++
|
|
if record.Late {
|
|
update.late++
|
|
}
|
|
if lastEventTime.IsZero() || record.EventAt.After(lastEventTime) {
|
|
lastEventTime = record.EventAt
|
|
update.lastEventAt = record.EventAt
|
|
update.lastReceivedAt = record.ReceivedAt
|
|
update.lastSourceEventID = record.SourceEventID
|
|
}
|
|
accepted = append(accepted, record)
|
|
} else {
|
|
update.invalid++
|
|
update.lastInvalidCode = record.ErrorCode
|
|
}
|
|
if record.Offset > maxOffset {
|
|
maxOffset = record.Offset
|
|
}
|
|
}
|
|
update.nextOffset = nextOffset
|
|
if maxOffset >= 0 && maxOffset+1 > update.nextOffset {
|
|
update.nextOffset = maxOffset + 1
|
|
}
|
|
if update.nextOffset < 0 {
|
|
update.nextOffset = 0
|
|
}
|
|
updates = append(updates, update)
|
|
result.Processed += int(update.processed)
|
|
result.Valid += int(update.valid)
|
|
result.Invalid += int(update.invalid)
|
|
result.Late += int(update.late)
|
|
result.ReplaySkipped += int(update.replay)
|
|
}
|
|
if strings.EqualFold(s.alertStreamMode, "active") && len(accepted) > 0 {
|
|
evaluation, evaluationErr := evaluateAlertStreamRecordsTx(ctx, tx, accepted, time.Now())
|
|
if evaluationErr != nil {
|
|
return result, evaluationErr
|
|
}
|
|
result.RulesEvaluated = evaluation.RulesEvaluated
|
|
result.CandidatesAdvanced = evaluation.CandidatesAdvanced
|
|
result.DuplicateObservations = evaluation.DuplicateObservations
|
|
result.LateObservations = evaluation.LateObservations
|
|
result.Opened = evaluation.Opened
|
|
result.Recovered = evaluation.Recovered
|
|
}
|
|
for _, update := range updates {
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO vehicle_alert_stream_checkpoint(consumer_group,topic,partition_id,next_offset,high_watermark,processed_count,valid_count,invalid_count,late_count,replay_skipped_count,last_event_at,last_received_at,last_source_event_id,last_invalid_code,last_invalid_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,IF(?='',NULL,CURRENT_TIMESTAMP(3))) ON DUPLICATE KEY UPDATE next_offset=GREATEST(next_offset,VALUES(next_offset)),high_watermark=GREATEST(high_watermark,VALUES(high_watermark)),processed_count=processed_count+VALUES(processed_count),valid_count=valid_count+VALUES(valid_count),invalid_count=invalid_count+VALUES(invalid_count),late_count=late_count+VALUES(late_count),replay_skipped_count=replay_skipped_count+VALUES(replay_skipped_count),last_event_at=COALESCE(VALUES(last_event_at),last_event_at),last_received_at=COALESCE(VALUES(last_received_at),last_received_at),last_source_event_id=IF(VALUES(last_source_event_id)='',last_source_event_id,VALUES(last_source_event_id)),last_invalid_code=IF(VALUES(last_invalid_code)='',last_invalid_code,VALUES(last_invalid_code)),last_invalid_at=COALESCE(VALUES(last_invalid_at),last_invalid_at)`, consumerGroup, update.key.topic, update.key.partition, update.nextOffset, update.highWatermark, update.processed, update.valid, update.invalid, update.late, update.replay, update.lastEventAt, update.lastReceivedAt, update.lastSourceEventID, update.lastInvalidCode, update.lastInvalidCode)
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
}
|
|
result.Partitions = len(keys)
|
|
if err = tx.Commit(); err != nil {
|
|
return result, err
|
|
}
|
|
return result, nil
|
|
}
|