From 009af2617ca5e45847a5b4aa0a8c308c8a70dbae Mon Sep 17 00:00:00 2001 From: JocLRojas Date: Tue, 21 Jul 2026 21:01:54 +0300 Subject: [PATCH 1/3] feat(plugins/enrichment): add CSV lookup enrichment plugin New Go plugin com.utmstack.enrichment that enriches log events by performing exact-match lookups against CSV files dropped in the filesystem at /workdir/pipeline/csv-datasets/. Invoked from filter YAMLs via the dynamic step. Filesystem is the source of truth: files are picked up automatically within 30s of being dropped, modified or deleted (polling watcher, no fsnotify to stay compatible with shared filesystems in multi-node deploys). Params for the dynamic step: dataset, source, match_column, output_column, destination. All required. Match values are normalised (lowercase + whitespace collapsed to underscore) both when indexing the CSV and when reading the source event field, so CSV maintainers and event producers can be inconsistent about case and spacing. Guardrails: 10000 rows hard limit per CSV (skip with ERROR), 5000 row warn threshold, UTF-8 required. Invalid or oversized files are skipped without affecting other datasets. Fail-open on any lookup miss (dataset not registered, column not present, no row match) so events keep flowing. Log dedup ensures warnings fire once per dataset id and errors once per (dataset, column) tuple to avoid log flood. Concurrency: registry uses atomic-swap pattern (readers hold an immutable *Dataset pointer). Per-column indices built lazily on first lookup and cached under a per-dataset RWMutex. --- plugins/enrichment/config/const.go | 14 ++ plugins/enrichment/go.mod | 58 ++++++ plugins/enrichment/go.sum | 161 +++++++++++++++ plugins/enrichment/internal/csvio/loader.go | 65 ++++++ .../enrichment/internal/csvio/separator.go | 36 ++++ .../enrichment/internal/parselog/parselog.go | 160 +++++++++++++++ .../enrichment/internal/registry/dataset.go | 19 ++ plugins/enrichment/internal/registry/dedup.go | 19 ++ plugins/enrichment/internal/registry/index.go | 33 ++++ .../enrichment/internal/registry/normalize.go | 9 + .../enrichment/internal/registry/registry.go | 48 +++++ plugins/enrichment/internal/storage/disk.go | 186 ++++++++++++++++++ .../enrichment/internal/watcher/watcher.go | 32 +++ plugins/enrichment/main.go | 37 ++++ 14 files changed, 877 insertions(+) create mode 100644 plugins/enrichment/config/const.go create mode 100644 plugins/enrichment/go.mod create mode 100644 plugins/enrichment/go.sum create mode 100644 plugins/enrichment/internal/csvio/loader.go create mode 100644 plugins/enrichment/internal/csvio/separator.go create mode 100644 plugins/enrichment/internal/parselog/parselog.go create mode 100644 plugins/enrichment/internal/registry/dataset.go create mode 100644 plugins/enrichment/internal/registry/dedup.go create mode 100644 plugins/enrichment/internal/registry/index.go create mode 100644 plugins/enrichment/internal/registry/normalize.go create mode 100644 plugins/enrichment/internal/registry/registry.go create mode 100644 plugins/enrichment/internal/storage/disk.go create mode 100644 plugins/enrichment/internal/watcher/watcher.go create mode 100644 plugins/enrichment/main.go diff --git a/plugins/enrichment/config/const.go b/plugins/enrichment/config/const.go new file mode 100644 index 000000000..ef36034ab --- /dev/null +++ b/plugins/enrichment/config/const.go @@ -0,0 +1,14 @@ +package config + +import "time" + +const ( + PluginName = "com.utmstack.enrichment" + ProcessName = "plugin_com.utmstack.enrichment" + + CSVSubdir = "csv-datasets" + + MaxCSVRows = 10000 + WarnCSVRows = 5000 + PollInterval = 30 * time.Second +) diff --git a/plugins/enrichment/go.mod b/plugins/enrichment/go.mod new file mode 100644 index 000000000..c14d5e688 --- /dev/null +++ b/plugins/enrichment/go.mod @@ -0,0 +1,58 @@ +module github.com/utmstack/UTMStack/plugins/enrichment + +go 1.25.5 + +require ( + github.com/threatwinds/go-sdk v1.1.26 + github.com/tidwall/gjson v1.19.0 + github.com/tidwall/sjson v1.2.5 + google.golang.org/protobuf v1.36.11 +) + +require ( + cel.dev/expr v0.25.2 // indirect + github.com/antlr4-go/antlr/v4 v4.13.1 // indirect + github.com/bytedance/gopkg v0.1.4 // indirect + github.com/bytedance/sonic v1.15.2 // indirect + github.com/bytedance/sonic/loader v0.5.1 // indirect + github.com/cloudwego/base64x v0.1.7 // indirect + github.com/gabriel-vasile/mimetype v1.4.13 // indirect + github.com/gin-contrib/sse v1.1.1 // indirect + github.com/gin-gonic/gin v1.12.0 // indirect + github.com/go-playground/locales v0.14.1 // indirect + github.com/go-playground/universal-translator v0.18.1 // indirect + github.com/go-playground/validator/v10 v10.30.3 // indirect + github.com/goccy/go-json v0.10.6 // indirect + github.com/goccy/go-yaml v1.19.2 // indirect + github.com/google/cel-go v0.28.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/klauspost/cpuid/v2 v2.3.0 // indirect + github.com/leodido/go-urn v1.4.0 // indirect + github.com/mattn/go-isatty v0.0.22 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/opensearch-project/opensearch-go/v4 v4.6.0 // indirect + github.com/pelletier/go-toml/v2 v2.3.1 // indirect + github.com/quic-go/qpack v0.6.0 // indirect + github.com/quic-go/quic-go v0.59.1 // indirect + github.com/tidwall/match v1.2.0 // indirect + github.com/tidwall/pretty v1.2.1 // indirect + github.com/twitchyliquid64/golang-asm v0.15.1 // indirect + github.com/ugorji/go/codec v1.3.1 // indirect + go.mongodb.org/mongo-driver/v2 v2.6.0 // indirect + go.yaml.in/yaml/v2 v2.4.4 // indirect + go.yaml.in/yaml/v3 v3.0.4 // indirect + golang.org/x/arch v0.27.0 // indirect + golang.org/x/crypto v0.52.0 // indirect + golang.org/x/exp v0.0.0-20260603202125-055de637280b // indirect + golang.org/x/net v0.55.0 // indirect + golang.org/x/sys v0.45.0 // indirect + golang.org/x/text v0.37.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/grpc v1.81.1 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect + sigs.k8s.io/yaml v1.6.0 // indirect +) diff --git a/plugins/enrichment/go.sum b/plugins/enrichment/go.sum new file mode 100644 index 000000000..ad2e2e9de --- /dev/null +++ b/plugins/enrichment/go.sum @@ -0,0 +1,161 @@ +cel.dev/expr v0.25.2 h1:K6j46C81hXtZQfuX60cVWQFBJahKSE2gfRbNuvr5bFs= +cel.dev/expr v0.25.2/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= +github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ= +github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw= +github.com/bytedance/gopkg v0.1.4 h1:oZnQwnX82KAIWb7033bEwtxvTqXcYMxDBaQxo5JJHWM= +github.com/bytedance/gopkg v0.1.4/go.mod h1:v1zWfPm21Fb+OsyXN2VAHdL6TBb2L88anLQgdyje6R4= +github.com/bytedance/sonic v1.15.2 h1:90H+rcF/FwLXwfB1cudOLq/je83n683Utf4Cbp0xHCo= +github.com/bytedance/sonic v1.15.2/go.mod h1:mT2NbXunuaEbnZ+mRIX/vYqKISmgEuHFDI4UzmKx2SA= +github.com/bytedance/sonic/loader v0.5.1 h1:Ygpfa9zwRCCKSlrp5bBP/b/Xzc3VxsAW+5NIYXrOOpI= +github.com/bytedance/sonic/loader v0.5.1/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/cloudwego/base64x v0.1.7 h1:NppS+Fgzg5ovhn4NkUXaDT3x9jldgH5ToMCqzBSi2zI= +github.com/cloudwego/base64x v0.1.7/go.mod h1:Cu1PV9zfrSf7ET2tIbWbbEy7jO7HHJ13q4X2SQ8aWYg= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/gabriel-vasile/mimetype v1.4.13 h1:46nXokslUBsAJE/wMsp5gtO500a4F3Nkz9Ufpk2AcUM= +github.com/gabriel-vasile/mimetype v1.4.13/go.mod h1:d+9Oxyo1wTzWdyVUPMmXFvp4F9tea18J8ufA774AB3s= +github.com/gin-contrib/sse v1.1.1 h1:uGYpNwTacv5R68bSGMapo62iLTRa9l5zxGCps4hK6ko= +github.com/gin-contrib/sse v1.1.1/go.mod h1:QXzuVkA0YO7o/gun03UI1Q+FTI8ZV/n5t03kIQAI89s= +github.com/gin-gonic/gin v1.12.0 h1:b3YAbrZtnf8N//yjKeU2+MQsh2mY5htkZidOM7O0wG8= +github.com/gin-gonic/gin v1.12.0/go.mod h1:VxccKfsSllpKshkBWgVgRniFFAzFb9csfngsqANjnLc= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= +github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= +github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA= +github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY= +github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY= +github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY= +github.com/go-playground/validator/v10 v10.30.3 h1:4MU6YkEwx7GbcPJOZxrtbu+QfF3pJLJuaYTeAH0DYy8= +github.com/go-playground/validator/v10 v10.30.3/go.mod h1:4Axh7oCNGcoGkqLoE4YWt6n20mcEIsPRlB7vPk3lpyc= +github.com/goccy/go-json v0.10.6 h1:p8HrPJzOakx/mn/bQtjgNjdTcN+/S6FcG2CTtQOrHVU= +github.com/goccy/go-json v0.10.6/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M= +github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM= +github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/cel-go v0.28.1 h1:YWIwi77J4xIsYUwAF/iIuS6haffzIHS8yWI8glSbLWM= +github.com/google/cel-go v0.28.1/go.mod h1:X0bD6iVNR8pkROSOoHVdgTkzmRcosof7WQqCD6wcMc8= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= +github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= +github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= +github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4= +github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/opensearch-project/opensearch-go/v4 v4.6.0 h1:Ac8aLtDSmLEyOmv0r1qhQLw3b4vcUhE42NE9k+Z4cRc= +github.com/opensearch-project/opensearch-go/v4 v4.6.0/go.mod h1:3iZtb4SNt3IzaxavKq0dURh1AmtVgYW71E4XqmYnIiQ= +github.com/pelletier/go-toml/v2 v2.3.1 h1:MYEvvGnQjeNkRF1qUuGolNtNExTDwct51yp7olPtrEc= +github.com/pelletier/go-toml/v2 v2.3.1/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= +github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= +github.com/quic-go/quic-go v0.59.1 h1:0Gmua0HW1Tv7ANR7hUYwRyD0MG5OJfgvYSZasGZzBic= +github.com/quic-go/quic-go v0.59.1/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/threatwinds/go-sdk v1.1.26 h1:9anBTRXXnNfft9FDgdasMOMUxtqlzE1Cm2b81lndFQQ= +github.com/threatwinds/go-sdk v1.1.26/go.mod h1:aN6Oe3zJop9ngS83oZcKFXDLKWzrny2XhkYm7uoyDbQ= +github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= +github.com/tidwall/gjson v1.19.0 h1:xwxm7n691Uf3u5OFjzngavjGTh55KX5q/9w9xHW88JU= +github.com/tidwall/gjson v1.19.0/go.mod h1:V37/opeE/JbLUOfH0QTXiNez2l0RUjYUhpT4szFQAfc= +github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM= +github.com/tidwall/match v1.2.0 h1:0pt8FlkOwjN2fPt4bIl4BoNxb98gGHN2ObFEDkrfZnM= +github.com/tidwall/match v1.2.0/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM= +github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4= +github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY= +github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28= +github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= +github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= +github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY= +github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4= +github.com/wI2L/jsondiff v0.7.0 h1:1lH1G37GhBPqCfp/lrs91rf/2j3DktX6qYAKZkLuCQQ= +github.com/wI2L/jsondiff v0.7.0/go.mod h1:KAEIojdQq66oJiHhDyQez2x+sRit0vIzC9KeK0yizxM= +go.mongodb.org/mongo-driver/v2 v2.6.0 h1:b9sJOYrkmt4l8bY43ZenFBcPlhYIjaOfYHLtbB/5qi8= +go.mongodb.org/mongo-driver/v2 v2.6.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= +go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/arch v0.27.0 h1:0WNVcR8u9yFz8j5FvdHpgwNp3FS5U4guYdzHwEiGjoU= +golang.org/x/arch v0.27.0/go.mod h1:0X+GdSIP+kL5wPmpK7sdkEVTt2XoYP0cSjQSbZBwOi8= +golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988= +golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= +golang.org/x/exp v0.0.0-20260603202125-055de637280b h1:v1uXiEBHo8QA0LiGCo7UgHMzHT4Kdfpl2zmtH5vaP1Q= +golang.org/x/exp v0.0.0-20260603202125-055de637280b/go.mod h1:d2fgXJLVs4dYDHUk5lwMIfzRzSrWCfGZb0ZqeLa/Vcw= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= +golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= +golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= +golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ= +google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs= +sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4= diff --git a/plugins/enrichment/internal/csvio/loader.go b/plugins/enrichment/internal/csvio/loader.go new file mode 100644 index 000000000..90ab7ebaa --- /dev/null +++ b/plugins/enrichment/internal/csvio/loader.go @@ -0,0 +1,65 @@ +package csvio + +import ( + "encoding/csv" + "errors" + "fmt" + "io" + "strings" + "time" + + "github.com/utmstack/UTMStack/plugins/enrichment/internal/registry" +) + +func LoadCSV(r io.Reader, separator rune, sizeBytes int64) (*registry.Dataset, error) { + reader := csv.NewReader(r) + reader.Comma = separator + reader.FieldsPerRecord = -1 + reader.LazyQuotes = true + + records, err := reader.ReadAll() + if err != nil { + return nil, fmt.Errorf("csv parse: %w", err) + } + + var nonEmpty [][]string + for _, rec := range records { + if len(rec) > 0 { + nonEmpty = append(nonEmpty, rec) + } + } + + if len(nonEmpty) < 1 { + return nil, errors.New("empty CSV: no header line found") + } + + headers := make([]string, len(nonEmpty[0])) + for i, h := range nonEmpty[0] { + headers[i] = strings.TrimSpace(h) + } + + colIndex := make(map[string]int, len(headers)) + for i, h := range headers { + colIndex[h] = i + } + + rawRows := make([][]string, 0, len(nonEmpty)-1) + for _, rec := range nonEmpty[1:] { + row := make([]string, len(rec)) + for i, v := range rec { + row[i] = strings.TrimSpace(v) + } + rawRows = append(rawRows, row) + } + + return ®istry.Dataset{ + Separator: separator, + Headers: headers, + ColIndex: colIndex, + RawRows: rawRows, + Indices: make(map[string]map[string][]int), + UploadedAt: time.Now().UTC(), + SizeBytes: sizeBytes, + RowCount: len(rawRows), + }, nil +} diff --git a/plugins/enrichment/internal/csvio/separator.go b/plugins/enrichment/internal/csvio/separator.go new file mode 100644 index 000000000..12d7cc9a6 --- /dev/null +++ b/plugins/enrichment/internal/csvio/separator.go @@ -0,0 +1,36 @@ +package csvio + +import "strings" + +func DetectSeparator(firstLine string) rune { + candidates := []rune{',', ';', '\t', '|'} + bestCount := 0 + bestSep := ',' + for _, c := range candidates { + count := strings.Count(firstLine, string(c)) + if count > bestCount { + bestCount = count + bestSep = c + } + } + return bestSep +} + +func ResolveSeparator(formValue string, content []byte) rune { + switch formValue { + case ",": + return ',' + case ";": + return ';' + case "\\t", "tab", "\t": + return '\t' + case "|": + return '|' + } + for _, line := range strings.Split(string(content), "\n") { + if strings.TrimSpace(line) != "" { + return DetectSeparator(line) + } + } + return ',' +} diff --git a/plugins/enrichment/internal/parselog/parselog.go b/plugins/enrichment/internal/parselog/parselog.go new file mode 100644 index 000000000..ad888bebf --- /dev/null +++ b/plugins/enrichment/internal/parselog/parselog.go @@ -0,0 +1,160 @@ +package parselog + +import ( + "context" + + "github.com/threatwinds/go-sdk/catcher" + "github.com/threatwinds/go-sdk/plugins" + "github.com/threatwinds/go-sdk/utils" + "github.com/tidwall/gjson" + "github.com/tidwall/sjson" + "google.golang.org/protobuf/types/known/structpb" + + "github.com/utmstack/UTMStack/plugins/enrichment/config" + "github.com/utmstack/UTMStack/plugins/enrichment/internal/registry" +) + +type parseParams struct { + dataset string + source string + matchColumn string + outputColumn string + destination string +} + +func ParseLog(_ context.Context, transform *plugins.Transform) (*plugins.Draft, error) { + params, err := extractParams(transform.Step.Dynamic.Params) + if err != nil { + return transform.Draft, err + } + + if err := validateFieldPaths(¶ms); err != nil { + return transform.Draft, err + } + + sourceValue := gjson.Get(transform.Draft.Log, params.source).String() + if sourceValue == "" { + return transform.Draft, nil + } + + outputValue, ok := lookupEnrichment(params, sourceValue) + if !ok { + return transform.Draft, nil + } + + return writeEnrichment(transform.Draft, params.destination, outputValue) +} + +func extractParams(raw map[string]*structpb.Value) (parseParams, error) { + dataset, err := requireString(raw, "dataset") + if err != nil { + return parseParams{}, err + } + source, err := requireString(raw, "source") + if err != nil { + return parseParams{}, err + } + matchColumn, err := requireString(raw, "match_column") + if err != nil { + return parseParams{}, err + } + outputColumn, err := requireString(raw, "output_column") + if err != nil { + return parseParams{}, err + } + destination, err := requireString(raw, "destination") + if err != nil { + return parseParams{}, err + } + return parseParams{ + dataset: dataset, + source: source, + matchColumn: matchColumn, + outputColumn: outputColumn, + destination: destination, + }, nil +} + +func requireString(raw map[string]*structpb.Value, key string) (string, error) { + v, ok := raw[key] + if !ok { + return "", catcher.Error("'"+key+"' param required", nil, procMeta()) + } + return v.GetStringValue(), nil +} + +func validateFieldPaths(p *parseParams) error { + utils.SanitizeField(&p.source) + if err := utils.ValidateReservedField(p.source, false); err != nil { + return catcher.Error("invalid source field", err, procMeta()) + } + utils.SanitizeField(&p.destination) + if err := utils.ValidateReservedField(p.destination, false); err != nil { + return catcher.Error("invalid destination field", err, procMeta()) + } + return nil +} + +func lookupEnrichment(p parseParams, sourceValue string) (string, bool) { + ds, ok := registry.Get(p.dataset) + if !ok { + if registry.MarkMissingDataset(p.dataset) { + catcher.Warn("dataset not found in registry (fail-open)", map[string]any{ + "process": config.ProcessName, + "datasetId": p.dataset, + }) + } + return "", false + } + + idx, err := ds.GetOrBuildIndex(p.matchColumn) + if err != nil { + if registry.MarkMissingColumn(p.dataset, p.matchColumn) { + _ = catcher.Error("match_column not in dataset", err, map[string]any{ + "process": config.ProcessName, + "datasetId": p.dataset, + "column": p.matchColumn, + }) + } + return "", false + } + + outputColIdx, ok := ds.ColIndex[p.outputColumn] + if !ok { + if registry.MarkMissingColumn(p.dataset, p.outputColumn) { + _ = catcher.Error("output_column not in dataset", nil, map[string]any{ + "process": config.ProcessName, + "datasetId": p.dataset, + "column": p.outputColumn, + }) + } + return "", false + } + + rowIndices, ok := idx[registry.NormalizeKey(sourceValue)] + if !ok || len(rowIndices) == 0 { + return "", false + } + + row := ds.RawRows[rowIndices[len(rowIndices)-1]] + if outputColIdx >= len(row) { + return "", false + } + return row[outputColIdx], true +} + +func writeEnrichment(draft *plugins.Draft, destination, value string) (*plugins.Draft, error) { + updatedLog, err := sjson.Set(draft.Log, destination, value) + if err != nil { + return draft, catcher.Error("failed to set enriched field", err, map[string]any{ + "process": config.ProcessName, + "destination": destination, + }) + } + draft.Log = updatedLog + return draft, nil +} + +func procMeta() map[string]any { + return map[string]any{"process": config.ProcessName} +} diff --git a/plugins/enrichment/internal/registry/dataset.go b/plugins/enrichment/internal/registry/dataset.go new file mode 100644 index 000000000..afeec7a7e --- /dev/null +++ b/plugins/enrichment/internal/registry/dataset.go @@ -0,0 +1,19 @@ +package registry + +import ( + "sync" + "time" +) + +type Dataset struct { + ID string + Separator rune + Headers []string + ColIndex map[string]int + RawRows [][]string + Indices map[string]map[string][]int + indicesMu sync.RWMutex + UploadedAt time.Time + SizeBytes int64 + RowCount int +} diff --git a/plugins/enrichment/internal/registry/dedup.go b/plugins/enrichment/internal/registry/dedup.go new file mode 100644 index 000000000..3ad5ef415 --- /dev/null +++ b/plugins/enrichment/internal/registry/dedup.go @@ -0,0 +1,19 @@ +package registry + +import "sync" + +var ( + missingDatasetSeen sync.Map + missingColumnSeen sync.Map +) + +func MarkMissingDataset(datasetID string) (firstTime bool) { + _, loaded := missingDatasetSeen.LoadOrStore(datasetID, struct{}{}) + return !loaded +} + +func MarkMissingColumn(datasetID, column string) (firstTime bool) { + key := datasetID + "|" + column + _, loaded := missingColumnSeen.LoadOrStore(key, struct{}{}) + return !loaded +} diff --git a/plugins/enrichment/internal/registry/index.go b/plugins/enrichment/internal/registry/index.go new file mode 100644 index 000000000..46e3f86a9 --- /dev/null +++ b/plugins/enrichment/internal/registry/index.go @@ -0,0 +1,33 @@ +package registry + +import "fmt" + +func (d *Dataset) GetOrBuildIndex(matchColumn string) (map[string][]int, error) { + d.indicesMu.RLock() + idx, ok := d.Indices[matchColumn] + d.indicesMu.RUnlock() + if ok { + return idx, nil + } + + colIdx, exists := d.ColIndex[matchColumn] + if !exists { + return nil, fmt.Errorf("column %q not in dataset headers", matchColumn) + } + + d.indicesMu.Lock() + defer d.indicesMu.Unlock() + if idx, ok = d.Indices[matchColumn]; ok { + return idx, nil + } + + newIdx := make(map[string][]int, len(d.RawRows)) + for i, row := range d.RawRows { + if colIdx < len(row) { + key := NormalizeKey(row[colIdx]) + newIdx[key] = append(newIdx[key], i) + } + } + d.Indices[matchColumn] = newIdx + return newIdx, nil +} diff --git a/plugins/enrichment/internal/registry/normalize.go b/plugins/enrichment/internal/registry/normalize.go new file mode 100644 index 000000000..2746ad636 --- /dev/null +++ b/plugins/enrichment/internal/registry/normalize.go @@ -0,0 +1,9 @@ +package registry + +import "strings" + +func NormalizeKey(s string) string { + lower := strings.ToLower(s) + parts := strings.Fields(lower) + return strings.Join(parts, "_") +} diff --git a/plugins/enrichment/internal/registry/registry.go b/plugins/enrichment/internal/registry/registry.go new file mode 100644 index 000000000..d05f69cab --- /dev/null +++ b/plugins/enrichment/internal/registry/registry.go @@ -0,0 +1,48 @@ +package registry + +import "sync" + +var ( + datasets = make(map[string]*Dataset) + registryMu sync.RWMutex +) + +func Get(id string) (*Dataset, bool) { + registryMu.RLock() + defer registryMu.RUnlock() + ds, ok := datasets[id] + return ds, ok +} + +func Swap(id string, ds *Dataset) { + registryMu.Lock() + defer registryMu.Unlock() + datasets[id] = ds +} + +func Delete(id string) bool { + registryMu.Lock() + defer registryMu.Unlock() + _, ok := datasets[id] + if ok { + delete(datasets, id) + } + return ok +} + +func Exists(id string) bool { + registryMu.RLock() + defer registryMu.RUnlock() + _, ok := datasets[id] + return ok +} + +func IDs() []string { + registryMu.RLock() + defer registryMu.RUnlock() + ids := make([]string, 0, len(datasets)) + for id := range datasets { + ids = append(ids, id) + } + return ids +} diff --git a/plugins/enrichment/internal/storage/disk.go b/plugins/enrichment/internal/storage/disk.go new file mode 100644 index 000000000..693fb9e9c --- /dev/null +++ b/plugins/enrichment/internal/storage/disk.go @@ -0,0 +1,186 @@ +package storage + +import ( + "bufio" + "bytes" + "fmt" + "os" + "strings" + "unicode/utf8" + + "github.com/threatwinds/go-sdk/catcher" + "github.com/threatwinds/go-sdk/plugins" + "github.com/threatwinds/go-sdk/utils" + + "github.com/utmstack/UTMStack/plugins/enrichment/config" + "github.com/utmstack/UTMStack/plugins/enrichment/internal/csvio" + "github.com/utmstack/UTMStack/plugins/enrichment/internal/registry" +) + +func DatasetsDir() (utils.Folder, error) { + return utils.MkdirJoin(plugins.WorkDir, "pipeline", config.CSVSubdir) +} + +func SyncFromDisk() error { + csvDir, err := DatasetsDir() + if err != nil { + return fmt.Errorf("could not create csv-datasets dir: %w", err) + } + + entries, err := os.ReadDir(csvDir.String()) + if err != nil { + if os.IsNotExist(err) { + removeVanishedDatasets(map[string]struct{}{}) + return nil + } + return err + } + + onDisk := make(map[string]struct{}, len(entries)) + loaded := 0 + skipped := 0 + + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".csv") { + continue + } + + id := strings.TrimSuffix(entry.Name(), ".csv") + filePath := csvDir.FileJoin(entry.Name()) + onDisk[id] = struct{}{} + + if !loadOrRefreshDataset(id, filePath) { + skipped++ + continue + } + loaded++ + } + + removed := removeVanishedDatasets(onDisk) + + catcher.Info("sync from disk complete", map[string]any{ + "process": config.ProcessName, + "loaded": loaded, + "skipped": skipped, + "removed": removed, + }) + return nil +} + +func loadOrRefreshDataset(id, filePath string) bool { + lineCount, err := countLines(filePath) + if err != nil { + catcher.Warn("could not count lines on csv — skipping", map[string]any{ + "process": config.ProcessName, + "path": filePath, + "id": id, + "error": err.Error(), + }) + return false + } + + if lineCount > config.MaxCSVRows+1 { + _ = catcher.Error("csv exceeds max rows limit — skipping", nil, map[string]any{ + "process": config.ProcessName, + "path": filePath, + "id": id, + "lines": lineCount, + "maxRows": config.MaxCSVRows, + }) + return false + } + + if lineCount > config.WarnCSVRows+1 { + catcher.Warn("large csv dataset", map[string]any{ + "process": config.ProcessName, + "id": id, + "lines": lineCount, + }) + } + + raw, err := os.ReadFile(filePath) + if err != nil { + catcher.Warn("could not read csv — skipping", map[string]any{ + "process": config.ProcessName, + "path": filePath, + "id": id, + "error": err.Error(), + }) + return false + } + + if !utf8.Valid(raw) { + catcher.Warn("invalid UTF-8 in csv — skipping", map[string]any{ + "process": config.ProcessName, + "path": filePath, + "id": id, + }) + return false + } + + var firstLine string + for _, line := range strings.Split(string(raw), "\n") { + if strings.TrimSpace(line) != "" { + firstLine = line + break + } + } + sep := csvio.DetectSeparator(firstLine) + + info, _ := os.Stat(filePath) + var sizeBytes int64 + if info != nil { + sizeBytes = info.Size() + } + + ds, err := csvio.LoadCSV(bytes.NewReader(raw), sep, sizeBytes) + if err != nil { + catcher.Warn("could not parse csv — skipping", map[string]any{ + "process": config.ProcessName, + "path": filePath, + "id": id, + "error": err.Error(), + }) + return false + } + ds.ID = id + + registry.Swap(id, ds) + return true +} + +func countLines(path string) (int, error) { + f, err := os.Open(path) + if err != nil { + return 0, err + } + defer f.Close() + + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 1024*1024), 1024*1024) + + count := 0 + for scanner.Scan() { + count++ + } + if err := scanner.Err(); err != nil { + return 0, err + } + return count, nil +} + +func removeVanishedDatasets(onDisk map[string]struct{}) int { + removed := 0 + for _, id := range registry.IDs() { + if _, exists := onDisk[id]; !exists { + if registry.Delete(id) { + catcher.Info("dataset removed (file no longer on disk)", map[string]any{ + "process": config.ProcessName, + "id": id, + }) + removed++ + } + } + } + return removed +} diff --git a/plugins/enrichment/internal/watcher/watcher.go b/plugins/enrichment/internal/watcher/watcher.go new file mode 100644 index 000000000..1570174d0 --- /dev/null +++ b/plugins/enrichment/internal/watcher/watcher.go @@ -0,0 +1,32 @@ +package watcher + +import ( + "context" + "time" + + "github.com/threatwinds/go-sdk/catcher" + + "github.com/utmstack/UTMStack/plugins/enrichment/config" + "github.com/utmstack/UTMStack/plugins/enrichment/internal/storage" +) + +func Start(ctx context.Context) { + ticker := time.NewTicker(config.PollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + catcher.Info("watcher stopped", map[string]any{ + "process": config.ProcessName, + }) + return + case <-ticker.C: + if err := storage.SyncFromDisk(); err != nil { + _ = catcher.Error("sync from disk failed", err, map[string]any{ + "process": config.ProcessName, + }) + } + } + } +} diff --git a/plugins/enrichment/main.go b/plugins/enrichment/main.go new file mode 100644 index 000000000..6c547bd4b --- /dev/null +++ b/plugins/enrichment/main.go @@ -0,0 +1,37 @@ +package main + +import ( + "context" + "os" + "time" + + "github.com/threatwinds/go-sdk/catcher" + "github.com/threatwinds/go-sdk/plugins" + + "github.com/utmstack/UTMStack/plugins/enrichment/config" + "github.com/utmstack/UTMStack/plugins/enrichment/internal/parselog" + "github.com/utmstack/UTMStack/plugins/enrichment/internal/storage" + "github.com/utmstack/UTMStack/plugins/enrichment/internal/watcher" +) + +func main() { + if err := storage.SyncFromDisk(); err != nil { + _ = catcher.Error("initial sync from disk failed", err, map[string]any{ + "process": config.ProcessName, + }) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go watcher.Start(ctx) + + err := plugins.InitParsingPlugin(config.PluginName, parselog.ParseLog) + if err != nil { + _ = catcher.Error("failed to init parsing plugin", err, map[string]any{ + "process": config.ProcessName, + }) + time.Sleep(5 * time.Second) + os.Exit(1) + } +} From 9412ec7c725b0db5a29ba47b063ca721272e2e9b Mon Sep 17 00:00:00 2001 From: JocLRojas Date: Tue, 21 Jul 2026 21:02:19 +0300 Subject: [PATCH 2/3] build(event-processor): include enrichment plugin in image Copies the com.utmstack.enrichment.plugin binary into /workdir/plugins/utmstack/ so the event-processor supervisor picks it up on container start and the parsing plugin exposes its Unix socket for the dynamic step. --- event_processor.Dockerfile | 1 + 1 file changed, 1 insertion(+) diff --git a/event_processor.Dockerfile b/event_processor.Dockerfile index b0dac424c..6b44b1fad 100644 --- a/event_processor.Dockerfile +++ b/event_processor.Dockerfile @@ -18,5 +18,6 @@ COPY ./plugins/stats/com.utmstack.stats.plugin /workdir/plugins/utmstack/ COPY ./plugins/soc-ai/com.utmstack.soc-ai.plugin /workdir/plugins/utmstack/ COPY ./plugins/modules-config/com.utmstack.modules-config.plugin /workdir/plugins/utmstack/ COPY ./plugins/crowdstrike/com.utmstack.crowdstrike.plugin /workdir/plugins/utmstack/ +COPY ./plugins/enrichment/com.utmstack.enrichment.plugin /workdir/plugins/utmstack/ COPY ./plugins/feeds/com.utmstack.feeds.plugin /workdir/plugins/utmstack/ COPY ./plugins/rule-flood-guard/com.utmstack.rule-flood-guard.plugin /workdir/plugins/utmstack/ \ No newline at end of file From 8072dc027fb723d5da18446e7e8396b9ad233473 Mon Sep 17 00:00:00 2001 From: JocLRojas Date: Tue, 21 Jul 2026 21:02:43 +0300 Subject: [PATCH 3/3] ci(v11): build enrichment plugin in deployment pipeline Adds the go build step for plugins/enrichment so the binary produced during the Build Plugins job is available for the event-processor Docker image COPY. --- .github/workflows/v11-deployment-pipeline.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/v11-deployment-pipeline.yml b/.github/workflows/v11-deployment-pipeline.yml index 985eba1e9..7621a901f 100644 --- a/.github/workflows/v11-deployment-pipeline.yml +++ b/.github/workflows/v11-deployment-pipeline.yml @@ -377,6 +377,7 @@ jobs: cd ${{ github.workspace }}/plugins/crowdstrike; go build -o com.utmstack.crowdstrike.plugin -v . cd ${{ github.workspace }}/plugins/feeds; go build -o com.utmstack.feeds.plugin -v . cd ${{ github.workspace }}/plugins/rule-flood-guard; go build -o com.utmstack.rule-flood-guard.plugin -v . + cd ${{ github.workspace }}/plugins/enrichment; go build -o com.utmstack.enrichment.plugin -v . - name: Prepare Dependencies for Event Processor Image run: |