gudae
2 months ago
10 changed files with 224 additions and 6 deletions
@ -0,0 +1,103 @@ |
|||||
|
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 |
||||
|
} |
||||
@ -0,0 +1,88 @@ |
|||||
|
package mock |
||||
|
|
||||
|
import ( |
||||
|
"context" |
||||
|
"testing" |
||||
|
"time" |
||||
|
|
||||
|
"tera-clone/services/tera-api/internal/domain" |
||||
|
"tera-clone/services/tera-api/internal/logstorage" |
||||
|
) |
||||
|
|
||||
|
func TestReceivedKeywordFromSerialJournalNotDependentOnLogViewer(t *testing.T) { |
||||
|
journal := &sequencedMemoryJournal{} |
||||
|
delivery := &recordingAlarmDeliveryController{} |
||||
|
store, err := NewStoreWithOptions(fixedClock, StoreOptions{LogJournal: journal, AlarmDeliveryController: delivery}) |
||||
|
if err != nil { |
||||
|
t.Fatal(err) |
||||
|
} |
||||
|
rule := store.runningAlarms["port_ch20_keyword"] |
||||
|
rule.Enabled = true |
||||
|
rule.Trigger = "Invalid interface" |
||||
|
rule.Delivery = []string{"web"} |
||||
|
store.runningAlarms["port_ch20_keyword"] = rule |
||||
|
now := fixedClock() |
||||
|
appendRX := func(channel, direction, text string, matched bool) { |
||||
|
t.Helper() |
||||
|
now = now.Add(time.Millisecond) |
||||
|
event := domain.PortLogEvent{Timestamp: now, Channel: channel, EventType: "traffic", Direction: direction, Actor: "tera-port-service", Message: "serial payload", Payload: []byte(text), PayloadBytes: len(text), PayloadEncoding: "base64", DisplayFormat: "ascii", AlarmTriggered: matched} |
||||
|
if err := journal.Append(logstorage.ClassPort, now, event); err != nil { |
||||
|
t.Fatal(err) |
||||
|
} |
||||
|
} |
||||
|
evaluate := func() { |
||||
|
t.Helper() |
||||
|
if err := store.EvaluateOperationalAlarms(context.Background()); err != nil { |
||||
|
t.Fatal(err) |
||||
|
} |
||||
|
} |
||||
|
appendRX("CH20", "RX", "Invalid interface", false) |
||||
|
evaluate() |
||||
|
if len(delivery.requests) != 0 { |
||||
|
t.Fatal("replayed old history on startup") |
||||
|
} |
||||
|
appendRX("CH20", "TX", "Invalid interface", false) |
||||
|
appendRX("CH19", "RX", "Invalid interface", false) |
||||
|
evaluate() |
||||
|
if len(delivery.requests) != 0 { |
||||
|
t.Fatal("matched wrong channel or TX") |
||||
|
} |
||||
|
appendRX("CH20", "RX", "\r\nInvalid inter", false) |
||||
|
evaluate() |
||||
|
if len(delivery.requests) != 0 { |
||||
|
t.Fatal("matched incomplete keyword") |
||||
|
} |
||||
|
appendRX("CH20", "RX", "face\r\nEnter interface name", false) |
||||
|
// A UI read advances its own journal cursor before the alarm evaluator.
|
||||
|
if _, err := store.PortLogEvents(context.Background(), domain.PortLogFilter{Channel: "CH20"}); err != nil { |
||||
|
t.Fatal(err) |
||||
|
} |
||||
|
evaluate() |
||||
|
evaluate() |
||||
|
if len(delivery.requests) != 1 || delivery.requests[0].RuleID != "port_ch20_keyword" { |
||||
|
t.Fatalf("requests=%+v", delivery.requests) |
||||
|
} |
||||
|
appendRX("CH20", "RX", "more prompt output", false) |
||||
|
appendRX("CH20", "RX", "Invalid interface", true) |
||||
|
evaluate() |
||||
|
if len(delivery.requests) != 1 { |
||||
|
t.Fatal("duplicate match or already processed API event") |
||||
|
} |
||||
|
rule.Enabled = false |
||||
|
store.runningAlarms["port_ch20_keyword"] = rule |
||||
|
appendRX("CH20", "RX", "Invalid interface", false) |
||||
|
evaluate() |
||||
|
if len(delivery.requests) != 1 { |
||||
|
t.Fatal("disabled rule fired") |
||||
|
} |
||||
|
rule.Enabled = true |
||||
|
rule.Trigger = "NEW KEYWORD" |
||||
|
store.runningAlarms["port_ch20_keyword"] = rule |
||||
|
appendRX("CH20", "RX", "Invalid interface", false) |
||||
|
evaluate() |
||||
|
appendRX("CH20", "RX", "new keyword", false) |
||||
|
evaluate() |
||||
|
if len(delivery.requests) != 2 { |
||||
|
t.Fatal("updated case insensitive keyword not observed") |
||||
|
} |
||||
|
} |
||||
Binary file not shown.
Loading…
Reference in new issue