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: | 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 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) + } +}