Skip to content

Commit ff25945

Browse files
ylladayllada
authored
feat(plugins): distribute ingestion ownership across event-processor workers (#2521)
* feat[plugins](shared): add NATS-based lease, cursor and job-queue coordination * feat[plugins](aws): own log groups by lease with per-stream cursors * feat[plugins](o365): consume ingestion jobs from the shared work queue * feat[plugins](sophos): consume ingestion jobs with a per-group cursor * feat[plugins](crowdstrike): own event streams by lease
1 parent f5cb983 commit ff25945

56 files changed

Lines changed: 2822 additions & 658 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎plugins/aws/client.go‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
package main
2+
3+
import (
4+
"sync"
5+
6+
"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
7+
8+
"github.com/utmstack/UTMStack/plugins/shared/identity"
9+
)
10+
11+
// clientCache hands out one client per (region, credentials). CloudWatch
12+
// Logs throttling quotas are enforced per account and region and shared by
13+
// every group using them, so client scope must match the quota, not the group.
14+
type clientCache struct {
15+
mu sync.Mutex
16+
clients map[string]*cloudwatchlogs.Client
17+
}
18+
19+
func newClientCache() *clientCache {
20+
return &clientCache{clients: make(map[string]*cloudwatchlogs.Client)}
21+
}
22+
23+
// Credentials are hashed because this key reaches log lines and error
24+
// contexts, where a secret access key must never appear.
25+
func clientCacheKey(region, accessKey, secretAccessKey string) string {
26+
return region + "|" + identity.Hash(accessKey, secretAccessKey)
27+
}
28+
29+
// The lock is intentionally not held across construction: it blocks on I/O
30+
// for up to a minute, so holding it would let one bad credential set stall
31+
// every other group's lookup. The re-check below discards a racing duplicate.
32+
func (c *clientCache) get(region, accessKey, secretAccessKey string) (*cloudwatchlogs.Client, error) {
33+
key := clientCacheKey(region, accessKey, secretAccessKey)
34+
35+
c.mu.Lock()
36+
cached, ok := c.clients[key]
37+
c.mu.Unlock()
38+
if ok {
39+
return cached, nil
40+
}
41+
42+
client, err := newCloudWatchLogsClient(region, accessKey, secretAccessKey)
43+
if err != nil {
44+
return nil, err
45+
}
46+
47+
c.mu.Lock()
48+
defer c.mu.Unlock()
49+
50+
if winner, ok := c.clients[key]; ok {
51+
return winner, nil
52+
}
53+
c.clients[key] = client
54+
return client, nil
55+
}
56+
57+
var awsClients = newClientCache()
58+
59+
func newCloudWatchLogsClient(region, accessKey, secretAccessKey string) (*cloudwatchlogs.Client, error) {
60+
processor := AWSProcessor{
61+
RegionName: region,
62+
AccessKey: accessKey,
63+
SecretAccessKey: secretAccessKey,
64+
}
65+
66+
cfg, err := processor.createAWSSession()
67+
if err != nil {
68+
return nil, err
69+
}
70+
71+
return cloudwatchlogs.NewFromConfig(cfg), nil
72+
}
73+
74+
func (p *AWSProcessor) client() (*cloudwatchlogs.Client, error) {
75+
return awsClients.get(p.RegionName, p.AccessKey, p.SecretAccessKey)
76+
}

‎plugins/aws/config.go‎

Lines changed: 6 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,6 @@
11
package main
22

33
import (
4-
"crypto/aes"
5-
"crypto/cipher"
6-
"crypto/sha1"
7-
"encoding/base64"
84
"fmt"
95
"os"
106
"path/filepath"
@@ -14,13 +10,14 @@ import (
1410
"github.com/fsnotify/fsnotify"
1511
"github.com/threatwinds/go-sdk/catcher"
1612
"github.com/threatwinds/go-sdk/plugins"
17-
"golang.org/x/crypto/pbkdf2"
13+
"github.com/utmstack/UTMStack/plugins/shared/crypto"
1814
"gopkg.in/yaml.v3"
1915
)
2016

2117
const (
2218
pluginFile = "system_plugins_aws.yaml"
23-
processName = "plugin_com.utmstack.aws"
19+
pluginName = "com.utmstack.aws"
20+
processName = "plugin_" + pluginName
2421
pipelineDirDefault = "/workdir/pipeline"
2522
)
2623

@@ -84,7 +81,6 @@ func StartConfigurationSystem() {
8481

8582
filePath := filepath.Join(pipelineDir, pluginFile)
8683

87-
// Initial load.
8884
if sec := readConfig(filePath, encKey); sec != nil {
8985
mu.Lock()
9086
cnf = sec
@@ -100,7 +96,7 @@ func StartConfigurationSystem() {
10096
}
10197
defer watcher.Close()
10298

103-
// Watch the directory so we catch atomic write (rename) events.
99+
// The directory, not the file: atomic writes arrive as renames.
104100
if err := watcher.Add(pipelineDir); err != nil {
105101
_ = catcher.Error("failed to watch pipeline dir", err, map[string]any{"process": processName})
106102
pollFallback(filePath, encKey)
@@ -182,8 +178,7 @@ func readConfig(path, encKey string) *ConfigurationSection {
182178
data, err := os.ReadFile(path)
183179
if err != nil {
184180
if os.IsNotExist(err) {
185-
// File removed → module disabled / no configuration. Report an empty,
186-
// inactive section so the module is treated as disabled and all work stops.
181+
// A removed file means the module is disabled, not an error.
187182
return &ConfigurationSection{ModuleActive: false}
188183
}
189184
_ = catcher.Error("failed to read config file", err, map[string]any{"process": processName, "file": path})
@@ -210,7 +205,7 @@ func readConfig(path, encKey string) *ConfigurationSection {
210205
for k, v := range g.Config {
211206
conf := &Configuration{ConfKey: k, ConfValue: v}
212207
if encKey != "" && sensitiveKeys[k] {
213-
dec, err := NewCipher(encKey).Decrypt(conf.ConfValue)
208+
dec, err := crypto.NewCipher(encKey).Decrypt(conf.ConfValue)
214209
if err == nil {
215210
conf.ConfValue = dec
216211
}
@@ -220,64 +215,6 @@ func readConfig(path, encKey string) *ConfigurationSection {
220215
sec.ModuleGroups = append(sec.ModuleGroups, grp)
221216
}
222217
}
223-
// A tenant section with no groups must not read as configured.
224218
sec.ModuleActive = len(sec.ModuleGroups) > 0
225219
return sec
226220
}
227-
228-
const (
229-
iterationCount = 65536
230-
keyLength = 16
231-
)
232-
233-
type Cipher struct {
234-
key []byte
235-
}
236-
237-
func NewCipher(key string) *Cipher {
238-
return &Cipher{key: []byte(key)}
239-
}
240-
241-
func (c *Cipher) setKey() (cipher.Block, []byte, error) {
242-
h := sha1.New()
243-
h.Write(c.key)
244-
salt := h.Sum(nil)
245-
keyEnc := pbkdf2.Key(c.key, salt, iterationCount, keyLength, sha1.New)
246-
block, err := aes.NewCipher(keyEnc)
247-
if err != nil {
248-
return nil, nil, err
249-
}
250-
return block, salt[:keyLength], nil
251-
}
252-
253-
func (c *Cipher) Decrypt(crypt string) (string, error) {
254-
if crypt == "" {
255-
return "", nil
256-
}
257-
encryptedData, err := base64.StdEncoding.DecodeString(crypt)
258-
if err != nil {
259-
return crypt, nil // not base64 → already plaintext
260-
}
261-
blk, iv, err := c.setKey()
262-
if err != nil {
263-
return crypt, err
264-
}
265-
if len(encryptedData)%aes.BlockSize != 0 {
266-
return crypt, nil // not a valid CBC block → already plaintext
267-
}
268-
dec := cipher.NewCBCDecrypter(blk, iv)
269-
decrypted := make([]byte, len(encryptedData))
270-
dec.CryptBlocks(decrypted, encryptedData)
271-
return string(pkcs5Trim(decrypted)), nil
272-
}
273-
274-
func pkcs5Trim(data []byte) []byte {
275-
if len(data) == 0 {
276-
return data
277-
}
278-
padding := int(data[len(data)-1])
279-
if padding > len(data) || padding == 0 {
280-
return data
281-
}
282-
return data[:len(data)-padding]
283-
}

‎plugins/aws/connectivity.go‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
package main
2+
3+
import (
4+
"context"
5+
"time"
6+
7+
"github.com/threatwinds/go-sdk/catcher"
8+
)
9+
10+
const connectivityRetryDelay = 5 * time.Second
11+
12+
func waitForConnectivity(ctx context.Context, checker func(string) error, url string, retryDelay time.Duration) {
13+
for {
14+
if err := checker(url); err != nil {
15+
_ = catcher.Error("failed to connect with external service", err, map[string]any{"process": processName})
16+
if !sleepWithCancel(ctx, retryDelay) {
17+
return
18+
}
19+
continue
20+
}
21+
return
22+
}
23+
}

‎plugins/aws/cursor.go‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
package main
2+
3+
import (
4+
"encoding/json"
5+
"sync"
6+
)
7+
8+
// One goroutine polls per log stream, so a group's position is the merge of
9+
// every stream's independently advancing NextForwardToken.
10+
type cursorMap struct {
11+
mu sync.Mutex
12+
entries map[string]*string
13+
}
14+
15+
func newCursorMap() *cursorMap {
16+
return &cursorMap{entries: make(map[string]*string)}
17+
}
18+
19+
// Only the stream's own goroutine may call this; a second writer for the
20+
// same key breaks delete's invariant.
21+
func (c *cursorMap) set(stream string, token *string) {
22+
c.mu.Lock()
23+
defer c.mu.Unlock()
24+
c.entries[stream] = token
25+
}
26+
27+
func (c *cursorMap) get(stream string) (*string, bool) {
28+
c.mu.Lock()
29+
defer c.mu.Unlock()
30+
v, ok := c.entries[stream]
31+
return v, ok
32+
}
33+
34+
// Callers must guarantee no set for this key can land afterwards, or the
35+
// entry is resurrected: cancel the stream's goroutine first, or call this
36+
// from that goroutine as it returns. To drop a live stream's token, use
37+
// set(stream, nil) instead.
38+
func (c *cursorMap) delete(stream string) {
39+
c.mu.Lock()
40+
defer c.mu.Unlock()
41+
delete(c.entries, stream)
42+
}
43+
44+
func (c *cursorMap) snapshot() map[string]*string {
45+
c.mu.Lock()
46+
defer c.mu.Unlock()
47+
out := make(map[string]*string, len(c.entries))
48+
for k, v := range c.entries {
49+
out[k] = v
50+
}
51+
return out
52+
}
53+
54+
func (c *cursorMap) marshalSnapshot() ([]byte, error) {
55+
return json.Marshal(c.snapshot())
56+
}
57+
58+
// Only safe before the group's stream goroutines start.
59+
func (c *cursorMap) replace(entries map[string]*string) {
60+
c.mu.Lock()
61+
defer c.mu.Unlock()
62+
if entries == nil {
63+
entries = make(map[string]*string)
64+
}
65+
c.entries = entries
66+
}
67+
68+
// A returned token makes GetLogEvents ignore StartTime. On nil the caller
69+
// must fall back to the group's baseline start time, not "now": streams are
70+
// only discovered every 5 minutes, so "now" silently drops everything a
71+
// late-discovered stream received before that.
72+
func seedNextToken(cursors *cursorMap, stream string) *string {
73+
if token, ok := cursors.get(stream); ok {
74+
return token
75+
}
76+
return nil
77+
}

‎plugins/aws/go.mod‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,14 +9,21 @@ require (
99
github.com/fsnotify/fsnotify v1.10.1
1010
github.com/google/uuid v1.6.0
1111
github.com/threatwinds/go-sdk v1.1.28
12+
github.com/utmstack/UTMStack/plugins/shared v0.0.0-00010101000000-000000000000
1213
)
1314

15+
replace github.com/utmstack/UTMStack/plugins/shared => ../shared
16+
1417
require (
1518
github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.38 // indirect
1619
github.com/aws/aws-sdk-go-v2/service/signin v1.5.6 // indirect
1720
github.com/bytedance/gopkg v0.1.4 // indirect
1821
github.com/goccy/go-yaml v1.19.2 // indirect
1922
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
23+
github.com/klauspost/compress v1.18.5 // indirect
24+
github.com/nats-io/nats.go v1.53.1 // indirect
25+
github.com/nats-io/nkeys v0.4.15 // indirect
26+
github.com/nats-io/nuid v1.0.1 // indirect
2027
github.com/opensearch-project/opensearch-go/v4 v4.6.0 // indirect
2128
github.com/quic-go/qpack v0.6.0 // indirect
2229
github.com/quic-go/quic-go v0.59.1 // indirect
@@ -63,7 +70,7 @@ require (
6370
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
6471
github.com/ugorji/go/codec v1.3.1 // indirect
6572
golang.org/x/arch v0.27.0 // indirect
66-
golang.org/x/crypto v0.55.0
73+
golang.org/x/crypto v0.55.0 // indirect
6774
golang.org/x/exp v0.0.0-20260603202125-055de637280b // indirect
6875
golang.org/x/net v0.57.0 // indirect
6976
golang.org/x/sys v0.47.0 // indirect

‎plugins/aws/go.sum‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,8 @@ github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs
8585
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
8686
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
8787
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
88+
github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
89+
github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
8890
github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
8991
github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
9092
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
@@ -100,6 +102,12 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w
100102
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
101103
github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
102104
github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
105+
github.com/nats-io/nats.go v1.53.1 h1:Otsq3uLc/kLdjmkNHkXH0jBqwUquwdKFoe3fq6/3/Xo=
106+
github.com/nats-io/nats.go v1.53.1/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
107+
github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
108+
github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
109+
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
110+
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
103111
github.com/opensearch-project/opensearch-go/v4 v4.6.0 h1:Ac8aLtDSmLEyOmv0r1qhQLw3b4vcUhE42NE9k+Z4cRc=
104112
github.com/opensearch-project/opensearch-go/v4 v4.6.0/go.mod h1:3iZtb4SNt3IzaxavKq0dURh1AmtVgYW71E4XqmYnIiQ=
105113
github.com/pelletier/go-toml/v2 v2.3.1 h1:MYEvvGnQjeNkRF1qUuGolNtNExTDwct51yp7olPtrEc=

0 commit comments

Comments
 (0)