You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
103 lines
3.1 KiB
103 lines
3.1 KiB
package mock
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"tera-clone/services/tera-api/internal/alarmdelivery"
|
|
"tera-clone/services/tera-api/internal/domain"
|
|
"tera-clone/services/tera-api/internal/logservice"
|
|
"tera-clone/services/tera-api/internal/logstorage"
|
|
)
|
|
|
|
type portKeywordTail struct {
|
|
trigger string
|
|
text string
|
|
at time.Time
|
|
}
|
|
|
|
// seriald writes RX directly to logd, bypassing RecordPortLog. Consume that
|
|
// journal with an independent cursor so viewing logs cannot consume alarms.
|
|
func (s *Store) evaluateReceivedPortKeywords(ctx context.Context, snapshot logstorage.Snapshot) error {
|
|
journal, ok := s.logJournal.(logservice.SequencedJournal)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
s.mu.Lock()
|
|
if !s.portKeywordStarted {
|
|
s.portKeywordStarted = true
|
|
s.portKeywordTails = make(map[string]portKeywordTail)
|
|
// Existing history must not be mailed again on service restart.
|
|
if next := snapshot.NextSequence[logstorage.ClassPort]; next > 0 {
|
|
s.portKeywordCursor = next - 1
|
|
s.mu.Unlock()
|
|
return nil
|
|
}
|
|
records, err := journal.ReadAfter(logstorage.ClassPort, 0)
|
|
if err != nil {
|
|
s.portKeywordStarted = false
|
|
s.mu.Unlock()
|
|
return err
|
|
}
|
|
for _, record := range records {
|
|
if record.Sequence > s.portKeywordCursor {
|
|
s.portKeywordCursor = record.Sequence
|
|
}
|
|
}
|
|
s.mu.Unlock()
|
|
return nil
|
|
}
|
|
records, err := journal.ReadAfter(logstorage.ClassPort, s.portKeywordCursor)
|
|
if err != nil {
|
|
s.mu.Unlock()
|
|
return fmt.Errorf("read serial keyword events: %w", err)
|
|
}
|
|
var requests []alarmdelivery.Request
|
|
for _, record := range records {
|
|
if record.Sequence <= s.portKeywordCursor {
|
|
continue
|
|
}
|
|
var event domain.PortLogEvent
|
|
if err := json.Unmarshal(record.Payload, &event); err != nil {
|
|
s.mu.Unlock()
|
|
return err
|
|
}
|
|
s.portKeywordCursor = record.Sequence
|
|
if event.EventType != "traffic" || event.Direction != "RX" || len(event.Payload) == 0 || event.AlarmTriggered {
|
|
continue
|
|
}
|
|
for id, rule := range s.runningAlarms {
|
|
if !rule.Enabled || rule.Source != "port_log" || rule.Trigger == "" || (rule.Channel != "" && rule.Channel != event.Channel) {
|
|
continue
|
|
}
|
|
key := id + ":" + event.Channel
|
|
tail := s.portKeywordTails[key]
|
|
if tail.trigger != rule.Trigger || event.Timestamp.Sub(tail.at) > 5*time.Second || event.Timestamp.Before(tail.at) {
|
|
tail.text = ""
|
|
}
|
|
content := tail.text + string(event.Payload)
|
|
if strings.Contains(strings.ToLower(content), strings.ToLower(rule.Trigger)) {
|
|
message := fmt.Sprintf("%s received keyword matched: %s", event.Channel, rule.Trigger)
|
|
requests = append(requests, s.alarmRequestsForRulesLocked([]string{id}, message)...)
|
|
}
|
|
// Keep only a suffix shorter than the keyword: an old match cannot fire
|
|
// on the following chunk, but a keyword split between reads still can.
|
|
keep := len(rule.Trigger) - 1
|
|
if keep > 1024 {
|
|
keep = 1024
|
|
}
|
|
if len(content) > keep {
|
|
content = content[len(content)-keep:]
|
|
}
|
|
s.portKeywordTails[key] = portKeywordTail{trigger: rule.Trigger, text: content, at: event.Timestamp}
|
|
}
|
|
}
|
|
s.mu.Unlock()
|
|
for i := range requests {
|
|
s.deliverBestEffort(ctx, &requests[i])
|
|
}
|
|
return nil
|
|
}
|
|
|