89 lines
2.2 KiB
Go
89 lines
2.2 KiB
Go
package processor
|
|
|
|
import (
|
|
"compress/gzip"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
"example.com/siem-greenfield/internal/clickhouse"
|
|
"example.com/siem-greenfield/internal/config"
|
|
"example.com/siem-greenfield/internal/contracts"
|
|
"example.com/siem-greenfield/internal/normalize"
|
|
"example.com/siem-greenfield/internal/queue"
|
|
)
|
|
|
|
func Run(ctx context.Context, cfg config.Config) error {
|
|
ch := clickhouse.New(cfg)
|
|
con := queue.NewConsumer(cfg.KafkaBrokers, cfg.KafkaTopic, cfg.KafkaGroup)
|
|
defer con.Close()
|
|
log.Printf("processor started topic=%s group=%s", cfg.KafkaTopic, cfg.KafkaGroup)
|
|
for {
|
|
m, e := con.Fetch(ctx)
|
|
if e != nil {
|
|
if ctx.Err() != nil {
|
|
return nil
|
|
}
|
|
return e
|
|
}
|
|
var env contracts.IngestEnvelope
|
|
if e = json.Unmarshal(m.Value, &env); e != nil {
|
|
log.Printf("dropping malformed queue message offset=%d: %v", m.Offset, e)
|
|
_ = con.Commit(ctx, m)
|
|
continue
|
|
}
|
|
rawKey := ""
|
|
if cfg.RawArchiveEnabled {
|
|
rawKey = normalize.ArchiveKey(env.ReceivedAt, env.AgentID, env.BatchUID, m.Partition, m.Offset)
|
|
if e = spool(cfg, rawKey, m.Value); e != nil {
|
|
log.Printf("raw spool failed offset=%d (message will be retried): %v", m.Offset, e)
|
|
time.Sleep(time.Second)
|
|
continue
|
|
}
|
|
}
|
|
events := make([]contracts.CanonicalEvent, 0, len(env.Events))
|
|
for i, p := range env.Events {
|
|
events = append(events, normalize.Event(env, p, rawKey, m.Partition, m.Offset, i))
|
|
}
|
|
cctx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
|
e = ch.InsertEvents(cctx, events)
|
|
cancel()
|
|
if e != nil {
|
|
log.Printf("clickhouse insert failed offset=%d: %v", m.Offset, e)
|
|
time.Sleep(time.Second)
|
|
continue
|
|
}
|
|
if e = con.Commit(ctx, m); e != nil {
|
|
return fmt.Errorf("commit queue offset: %w", e)
|
|
}
|
|
}
|
|
}
|
|
func spool(cfg config.Config, key string, payload []byte) error {
|
|
path := filepath.Join(cfg.RawSpoolDir, filepath.FromSlash(key))
|
|
if e := os.MkdirAll(filepath.Dir(path), 0750); e != nil {
|
|
return e
|
|
}
|
|
tmp := path + ".tmp"
|
|
f, e := os.Create(tmp)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
gz := gzip.NewWriter(f)
|
|
_, e = gz.Write(payload)
|
|
if ce := gz.Close(); e == nil {
|
|
e = ce
|
|
}
|
|
if ce := f.Close(); e == nil {
|
|
e = ce
|
|
}
|
|
if e != nil {
|
|
_ = os.Remove(tmp)
|
|
return e
|
|
}
|
|
return os.Rename(tmp, path)
|
|
}
|