Skip to content

Commit 9715264

Browse files
AlexSanchez-bitAlexSanchez-bit
authored
fix[backend](integrations): fixed plugins yaml files integration form… (#2341)
* fix[backend](integrations): fixed plugins yaml files integration format to match event processor expected format * fix[backend](integrations): added lock files to avoid data duplication or ccorruption * fix[backend](integrations): used flock to handle locks with kernel locking
1 parent 0cb33f5 commit 9715264

2 files changed

Lines changed: 203 additions & 39 deletions

File tree

‎backend/modules/integrations/repository/tenant_store.go‎

Lines changed: 88 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -7,12 +7,32 @@ import (
77
"path/filepath"
88
"strings"
99
"sync"
10+
"syscall"
1011

1112
"github.com/utmstack/utmstack/backend/modules/integrations/connectors"
1213
"github.com/utmstack/utmstack/backend/modules/integrations/domain"
1314
"gopkg.in/yaml.v3"
1415
)
1516

17+
// withFileLock acquires an exclusive advisory lock on <path>.lock via flock,
18+
// runs fn, releases the lock on close. The kernel drops the lock on fd close
19+
// (including process death), so a crashed writer never leaves a stuck lock.
20+
// Linux/BSD only; UTMStack runs in Linux containers.
21+
func withFileLock(path string, fn func() error) error {
22+
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
23+
return err
24+
}
25+
lf, err := os.OpenFile(path+".lock", os.O_CREATE|os.O_RDWR, 0o600)
26+
if err != nil {
27+
return err
28+
}
29+
defer lf.Close()
30+
if err := syscall.Flock(int(lf.Fd()), syscall.LOCK_EX); err != nil {
31+
return err
32+
}
33+
return fn()
34+
}
35+
1636
var _ connectors.TenantRepository = (*TenantStore)(nil)
1737

1838
type TenantStore struct {
@@ -37,29 +57,44 @@ var pluginNames = map[string]string{
3757
"SOPHOS": "sophos",
3858
}
3959

40-
func tenantFileName(module string) string {
60+
func pluginKey(module string) string {
4161
name, ok := pluginNames[strings.ToUpper(module)]
4262
if !ok {
4363
name = strings.ToLower(module)
4464
}
45-
return "system_plugins_" + name + ".yaml"
65+
return name
66+
}
67+
68+
func tenantFileName(module string) string {
69+
return "system_plugins_" + pluginKey(module) + ".yaml"
4670
}
4771

4872
func (s *TenantStore) path(module string) string {
4973
return filepath.Join(s.dir, tenantFileName(module))
5074
}
5175

76+
// pluginsFile is the on-disk wrapper: plugins.<name>.tenants: [...].
77+
type pluginsFile struct {
78+
Plugins map[string]pluginEntry `yaml:"plugins"`
79+
}
80+
81+
type pluginEntry struct {
82+
Tenants []domain.Tenant `yaml:"tenants"`
83+
}
84+
5285
func (s *TenantStore) SetActiveByModule(_ context.Context, module string, active bool) error {
5386
if active {
5487
return nil
5588
}
5689
s.mu.Lock()
5790
defer s.mu.Unlock()
5891

59-
if err := os.Remove(s.path(module)); err != nil && !errors.Is(err, os.ErrNotExist) {
60-
return err
61-
}
62-
return nil
92+
return withFileLock(s.path(module), func() error {
93+
if err := os.Remove(s.path(module)); err != nil && !errors.Is(err, os.ErrNotExist) {
94+
return err
95+
}
96+
return nil
97+
})
6398
}
6499

65100
func (s *TenantStore) Load(module string) ([]domain.Tenant, error) {
@@ -70,18 +105,28 @@ func (s *TenantStore) Load(module string) ([]domain.Tenant, error) {
70105
if err != nil {
71106
return nil, err
72107
}
73-
var tenants []domain.Tenant
74-
if err := yaml.Unmarshal(data, &tenants); err != nil {
108+
var pf pluginsFile
109+
if err := yaml.Unmarshal(data, &pf); err != nil {
75110
return nil, err
76111
}
77-
return tenants, nil
112+
entry, ok := pf.Plugins[pluginKey(module)]
113+
if !ok {
114+
return []domain.Tenant{}, nil
115+
}
116+
return entry.Tenants, nil
78117
}
79118

80119
func (s *TenantStore) Save(module string, tenants []domain.Tenant) error {
81-
if err := os.MkdirAll(s.dir, 0o755); err != nil {
82-
return err
83-
}
84-
data, err := yaml.Marshal(tenants)
120+
return withFileLock(s.path(module), func() error {
121+
return s.saveLocked(module, tenants)
122+
})
123+
}
124+
125+
// saveLocked writes the file assuming the caller already holds the file lock.
126+
func (s *TenantStore) saveLocked(module string, tenants []domain.Tenant) error {
127+
key := pluginKey(module)
128+
pf := pluginsFile{Plugins: map[string]pluginEntry{key: {Tenants: tenants}}}
129+
data, err := yaml.Marshal(pf)
85130
if err != nil {
86131
return err
87132
}
@@ -97,38 +142,42 @@ func (s *TenantStore) Upsert(module string, t domain.Tenant) error {
97142
s.mu.Lock()
98143
defer s.mu.Unlock()
99144

100-
tenants, err := s.Load(module)
101-
if err != nil {
102-
return err
103-
}
104-
for i := range tenants {
105-
if tenants[i].Name == t.Name {
106-
tenants[i] = t
107-
return s.Save(module, tenants)
145+
return withFileLock(s.path(module), func() error {
146+
tenants, err := s.Load(module)
147+
if err != nil {
148+
return err
108149
}
109-
}
110-
return s.Save(module, append(tenants, t))
150+
for i := range tenants {
151+
if tenants[i].Name == t.Name {
152+
tenants[i] = t
153+
return s.saveLocked(module, tenants)
154+
}
155+
}
156+
return s.saveLocked(module, append(tenants, t))
157+
})
111158
}
112159

113160
func (s *TenantStore) Delete(module, name string) error {
114161
s.mu.Lock()
115162
defer s.mu.Unlock()
116163

117-
tenants, err := s.Load(module)
118-
if err != nil {
119-
return err
120-
}
121-
out := make([]domain.Tenant, 0, len(tenants))
122-
found := false
123-
for _, t := range tenants {
124-
if t.Name == name {
125-
found = true
126-
continue
164+
return withFileLock(s.path(module), func() error {
165+
tenants, err := s.Load(module)
166+
if err != nil {
167+
return err
127168
}
128-
out = append(out, t)
129-
}
130-
if !found {
131-
return domain.ErrTenantNotFound
132-
}
133-
return s.Save(module, out)
169+
out := make([]domain.Tenant, 0, len(tenants))
170+
found := false
171+
for _, t := range tenants {
172+
if t.Name == name {
173+
found = true
174+
continue
175+
}
176+
out = append(out, t)
177+
}
178+
if !found {
179+
return domain.ErrTenantNotFound
180+
}
181+
return s.saveLocked(module, out)
182+
})
134183
}

‎backend/modules/integrations/repository/tenant_store_test.go‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
package repository
2+
3+
import (
4+
"os"
5+
"strings"
6+
"syscall"
7+
"testing"
8+
"time"
9+
10+
"github.com/utmstack/utmstack/backend/modules/integrations/domain"
11+
)
12+
13+
func TestTenantStoreRoundtrip(t *testing.T) {
14+
dir := t.TempDir()
15+
s := NewTenantStore(dir)
16+
17+
want := []domain.Tenant{
18+
{Name: "t1", Config: map[string]string{"access_key": "AK", "secret": "SK"}},
19+
{Name: "t2", Config: map[string]string{"access_key": "AK2"}},
20+
}
21+
if err := s.Save("AWS_IAM_USER", want); err != nil {
22+
t.Fatalf("Save: %v", err)
23+
}
24+
got, err := s.Load("AWS_IAM_USER")
25+
if err != nil {
26+
t.Fatalf("Load: %v", err)
27+
}
28+
if len(got) != len(want) {
29+
t.Fatalf("len(got)=%d want %d", len(got), len(want))
30+
}
31+
for i := range want {
32+
if got[i].Name != want[i].Name {
33+
t.Fatalf("[%d] name got %q want %q", i, got[i].Name, want[i].Name)
34+
}
35+
for k, v := range want[i].Config {
36+
if got[i].Config[k] != v {
37+
t.Fatalf("[%d] %s got %q want %q", i, k, got[i].Config[k], v)
38+
}
39+
}
40+
}
41+
42+
data, err := os.ReadFile(s.path("AWS_IAM_USER"))
43+
if err != nil {
44+
t.Fatalf("read: %v", err)
45+
}
46+
txt := string(data)
47+
if !strings.Contains(txt, "plugins:") || !strings.Contains(txt, "aws:") || !strings.Contains(txt, "tenants:") {
48+
t.Fatalf("unexpected file layout:\n%s", txt)
49+
}
50+
}
51+
52+
func TestWithFileLockWaits(t *testing.T) {
53+
dir := t.TempDir()
54+
target := dir + "/f.yaml"
55+
56+
// pre-hold the flock via a peer fd
57+
peer, err := os.OpenFile(target+".lock", os.O_CREATE|os.O_RDWR, 0o600)
58+
if err != nil {
59+
t.Fatalf("open: %v", err)
60+
}
61+
if err := syscall.Flock(int(peer.Fd()), syscall.LOCK_EX); err != nil {
62+
t.Fatalf("flock: %v", err)
63+
}
64+
65+
done := make(chan error, 1)
66+
go func() {
67+
done <- withFileLock(target, func() error { return nil })
68+
}()
69+
70+
select {
71+
case <-done:
72+
t.Fatal("withFileLock returned while flock was held")
73+
case <-time.After(120 * time.Millisecond):
74+
}
75+
76+
if err := syscall.Flock(int(peer.Fd()), syscall.LOCK_UN); err != nil {
77+
t.Fatalf("unlock: %v", err)
78+
}
79+
_ = peer.Close()
80+
81+
select {
82+
case err := <-done:
83+
if err != nil {
84+
t.Fatalf("withFileLock: %v", err)
85+
}
86+
case <-time.After(500 * time.Millisecond):
87+
t.Fatal("withFileLock did not return after flock released")
88+
}
89+
}
90+
91+
// TestWithFileLockKernelReleasesOnClose proves the crash-recovery property: if
92+
// the peer holding the flock closes its fd (equivalent to process death), the
93+
// kernel releases the lock and a new acquirer proceeds without any TTL/steal.
94+
func TestWithFileLockKernelReleasesOnClose(t *testing.T) {
95+
dir := t.TempDir()
96+
target := dir + "/f.yaml"
97+
98+
peer, err := os.OpenFile(target+".lock", os.O_CREATE|os.O_RDWR, 0o600)
99+
if err != nil {
100+
t.Fatalf("open: %v", err)
101+
}
102+
if err := syscall.Flock(int(peer.Fd()), syscall.LOCK_EX); err != nil {
103+
t.Fatalf("flock: %v", err)
104+
}
105+
// simulate crash: close fd without explicit unlock
106+
_ = peer.Close()
107+
108+
start := time.Now()
109+
if err := withFileLock(target, func() error { return nil }); err != nil {
110+
t.Fatalf("withFileLock: %v", err)
111+
}
112+
if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
113+
t.Fatalf("lock not released on close: %v", elapsed)
114+
}
115+
}

0 commit comments

Comments
 (0)