multiindex search for elastiwatch (#365)

* multiindex search for elastiwatch

* x

Author: zoom-ua

elastiwatch/elastiwatch/entry_test.go

@@ -0,0 +1,36 @@
+package elastiwatch
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewEntryGregor(t *testing.T) {
+	e, err := newEntry([]byte(`{"message":"boom","severity":"ERROR","program":"dbcleand","time":"2026-10-22T15:00:00Z"}`))
+	require.NoError(t, err)
+	require.Equal(t, "boom", e.Message)
+	require.Equal(t, "ERROR", e.Severity)
+	require.Equal(t, "dbcleand", e.Program)
+}
+
+func TestNewEntryKBFSLogrus(t *testing.T) {
+	e, err := newEntry([]byte(`{"msg":"block get failed","level":"error","time":"2026-10-22T15:00:00.123456789Z"}`))
+	require.NoError(t, err)
+	require.Equal(t, "block get failed", e.Message)
+	require.Equal(t, "ERROR", e.Severity)
+}
+
+func TestNewEntryIced(t *testing.T) {
+	e, err := newEntry([]byte(`{"message":"node version: v22.0.0","level":"info","program":"logdumpd","pid":13,"@timestamp":"2026-10-22T15:00:00.000Z"}`))
+	require.NoError(t, err)
+	require.Equal(t, "node version: v22.0.0", e.Message)
+	require.Equal(t, "INFO", e.Severity)
+	require.Equal(t, "logdumpd", e.Program)
+	require.Equal(t, "2026-10-22T15:00:00.000Z", e.Time)
+}
+
+func TestNormalizeSeverityWarnFatal(t *testing.T) {
+	require.Equal(t, "WARNING", normalizeSeverity("warn"))
+	require.Equal(t, "CRITICAL", normalizeSeverity("fatal"))
+}

elastiwatch/elastiwatch/grouper.go

@@ -9,19 +9,49 @@ import (
 )
 
 type entry struct {
-	Message  string `json:"message"`
-	Severity string `json:"severity"`
-	UID      string `json:"uid"`
-	Time     string `json:"time"`
-	Program  string `json:"program"`
-	parts    []string
+	Message   string `json:"message"`
+	Msg       string `json:"msg"`
+	Severity  string `json:"severity"`
+	Level     string `json:"level"`
+	UID       string `json:"uid"`
+	Time      string `json:"time"`
+	Timestamp string `json:"@timestamp"`
+	Program   string `json:"program"`
+	parts     []string
+}
+
+func normalizeSeverity(s string) string {
+	switch strings.ToUpper(strings.TrimSpace(s)) {
+	case "WARN", "WARNING":
+		return "WARNING"
+	case "ERR", "ERROR":
+		return "ERROR"
+	case "FATAL", "PANIC", "CRIT", "CRITICAL":
+		return "CRITICAL"
+	case "INFO", "INFORMATION", "NOTICE":
+		return "INFO"
+	case "DEBUG", "TRACE":
+		return "DEBUG"
+	default:
+		return strings.ToUpper(strings.TrimSpace(s))
+	}
 }
 
 func newEntry(dat []byte) (*entry, error) {
 	entry := new(entry)
 	if err := json.Unmarshal(dat, entry); err != nil {
 		return nil, err
 	}
+	if entry.Message == "" {
+		entry.Message = entry.Msg
+	}
+	if entry.Severity == "" {
+		entry.Severity = entry.Level
+	}
+	entry.Severity = normalizeSeverity(entry.Severity)
+	if entry.Time == "" {
+		entry.Time = entry.Timestamp
+	}
 	entry.parts = append(entry.parts, entry.Program)
 	entry.parts = append(entry.parts, entry.UID)
 	entry.parts = append(entry.parts, strings.Split(entry.Message, " ")...)

elastiwatch/elastiwatch/logs.go

@@ -6,6 +6,7 @@ import (
 	"regexp"
 	"strings"
 	"time"
+	"unicode"
 
 	"github.com/keybase/go-keybase-chat-bot/kbchat/types/chat1"
 	"github.com/keybase/managed-bots/base"
@@ -14,25 +15,28 @@ import (
 
 type LogWatch struct {
 	*base.DebugOutput
-	db           *DB
-	cli          *opensearchapi.Client
-	index, email string
-	entries      []*entry
-	emailer      base.Emailer
-	sendCount    int
-	lastSend     time.Time
-	alertConvID  chat1.ConvIDStr
-	emailConvID  chat1.ConvIDStr
-	shutdownCh   chan struct{}
-	peekCh       chan struct{}
+	db          *DB
+	cli         *opensearchapi.Client
+	indices     []string
+	email       string
+	entries     []*entry
+	emailer     base.Emailer
+	sendCount   int
+	lastSend    time.Time
+	alertConvID chat1.ConvIDStr
+	emailConvID chat1.ConvIDStr
+	shutdownCh  chan struct{}
+	peekCh      chan struct{}
 }
 
 func NewLogWatch(cli *opensearchapi.Client, db *DB, index, email string, emailer base.Emailer, alertConvID, emailConvID chat1.ConvIDStr, debugConfig *base.ChatDebugOutputConfig) *LogWatch {
 	return &LogWatch{
 		DebugOutput: base.NewDebugOutput("LogWatch", debugConfig),
 		cli:         cli,
 		db:          db,
-		index:       index,
+		indices: strings.FieldsFunc(index, func(r rune) bool {
+			return r == ',' || unicode.IsSpace(r)
+		}),
 		email:       email,
 		emailer:     emailer,
 		lastSend:    time.Now(),
@@ -168,9 +172,9 @@ func (l *LogWatch) generateAndSend(entries []*entry) {
 
 func (l *LogWatch) runOnce() {
 	res, err := l.cli.Search(context.Background(), &opensearchapi.SearchReq{
-		Indices: []string{l.index},
+		Indices: l.indices,
 		Params: opensearchapi.SearchParams{
-			Query:  `NOT severity:debug AND @timestamp:[now-1m TO now]`,
+			Query:  `NOT severity:debug AND NOT level:debug AND @timestamp:[now-1m TO now]`,
 			Sort:   []string{"@timestamp:desc"},
 			Size:   new(10000),
 			Pretty: true,
@@ -190,6 +194,9 @@ func (l *LogWatch) runOnce() {
 				l.Errorf("failed to unmarshal log entry: %s", err)
 				continue
 			}
+			if entry.Severity == "DEBUG" {
+				continue
+			}
 			entries = append(entries, entry)
 		}
 	} else {

elastiwatch/main.go

@@ -176,7 +176,7 @@ func mainInner() int {
 	fs := flag.NewFlagSet(os.Args[0], flag.ExitOnError)
 	var alertConvID, emailConvID string
 	fs.StringVar(&opts.ESAddress, "esaddress", os.Getenv("ESADDRESS"), "Elasticsearch address")
-	fs.StringVar(&opts.Index, "index", os.Getenv("INDEX"), "Elasticsearch index")
+	fs.StringVar(&opts.Index, "index", os.Getenv("INDEX"), "OpenSearch index pattern(s), comma-separated")
 	fs.StringVar(&opts.Email, "email", os.Getenv("EMAIL"), "Destination email address")
 	fs.StringVar(&opts.SenderEmail, "sender-email", os.Getenv("SENDER_EMAIL"), "Sourceemail address")
 	fs.StringVar(&opts.Team, "team", os.Getenv("TEAM"), "Team")