Chore: Fix plugins manager process data race in tests (#81914)

* Chore: synchronize writes to pkg.plugins.log.Logs to prevent data races in test code

* Chore: fix data race in tests in plugins process manager

* Chore: improve Logs method naming

* Chore: fix type change
pull/81963/head^2
Diego Augusto Molina 2 years ago committed by GitHub
parent 1b63c27bb4
commit a6e27d1622
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
  1. 31
      pkg/plugins/log/fake.go
  2. 24
      pkg/plugins/manager/process/process.go
  3. 51
      pkg/plugins/manager/process/process_test.go

@ -1,6 +1,9 @@
package log package log
import "context" import (
"context"
"sync"
)
var _ Logger = (*TestLogger)(nil) var _ Logger = (*TestLogger)(nil)
@ -20,27 +23,19 @@ func (f *TestLogger) New(_ ...any) Logger {
} }
func (f *TestLogger) Info(msg string, ctx ...any) { func (f *TestLogger) Info(msg string, ctx ...any) {
f.InfoLogs.Calls++ f.InfoLogs.Call(msg, ctx)
f.InfoLogs.Message = msg
f.InfoLogs.Ctx = ctx
} }
func (f *TestLogger) Warn(msg string, ctx ...any) { func (f *TestLogger) Warn(msg string, ctx ...any) {
f.WarnLogs.Calls++ f.WarnLogs.Call(msg, ctx)
f.WarnLogs.Message = msg
f.WarnLogs.Ctx = ctx
} }
func (f *TestLogger) Debug(msg string, ctx ...any) { func (f *TestLogger) Debug(msg string, ctx ...any) {
f.DebugLogs.Calls++ f.DebugLogs.Call(msg, ctx)
f.DebugLogs.Message = msg
f.DebugLogs.Ctx = ctx
} }
func (f *TestLogger) Error(msg string, ctx ...any) { func (f *TestLogger) Error(msg string, ctx ...any) {
f.ErrorLogs.Calls++ f.ErrorLogs.Call(msg, ctx)
f.ErrorLogs.Message = msg
f.ErrorLogs.Ctx = ctx
} }
func (f *TestLogger) FromContext(_ context.Context) Logger { func (f *TestLogger) FromContext(_ context.Context) Logger {
@ -51,6 +46,16 @@ type Logs struct {
Calls int Calls int
Message string Message string
Ctx []any Ctx []any
mu sync.Mutex
}
func (l *Logs) Call(msg string, ctx ...any) {
l.mu.Lock()
defer l.mu.Unlock()
l.Calls++
l.Message = msg
l.Ctx = ctx
} }
var _ PrettyLogger = (*TestPrettyLogger)(nil) var _ PrettyLogger = (*TestPrettyLogger)(nil)

@ -7,22 +7,24 @@ import (
"github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/plugins"
) )
var ( const defaultKeepPluginAliveTickerDuration = time.Second
keepPluginAliveTickerDuration = time.Second * 1
)
type Service struct{} type Service struct {
keepPluginAliveTickerDuration time.Duration
}
func ProvideService() *Service { func ProvideService() *Service {
return &Service{} return &Service{
keepPluginAliveTickerDuration: defaultKeepPluginAliveTickerDuration,
}
} }
func (*Service) Start(ctx context.Context, p *plugins.Plugin) error { func (s *Service) Start(ctx context.Context, p *plugins.Plugin) error {
if !p.IsManaged() || !p.Backend || p.SignatureError != nil { if !p.IsManaged() || !p.Backend || p.SignatureError != nil {
return nil return nil
} }
if err := startPluginAndKeepItAlive(ctx, p); err != nil { if err := s.startPluginAndKeepItAlive(ctx, p); err != nil {
return err return err
} }
@ -43,7 +45,7 @@ func (*Service) Stop(ctx context.Context, p *plugins.Plugin) error {
return nil return nil
} }
func startPluginAndKeepItAlive(ctx context.Context, p *plugins.Plugin) error { func (s *Service) startPluginAndKeepItAlive(ctx context.Context, p *plugins.Plugin) error {
if err := p.Start(ctx); err != nil { if err := p.Start(ctx); err != nil {
return err return err
} }
@ -53,7 +55,7 @@ func startPluginAndKeepItAlive(ctx context.Context, p *plugins.Plugin) error {
} }
go func(p *plugins.Plugin) { go func(p *plugins.Plugin) {
if err := keepPluginAlive(p); err != nil { if err := s.keepPluginAlive(p); err != nil {
p.Logger().Error("Attempt to restart killed plugin process failed", "error", err) p.Logger().Error("Attempt to restart killed plugin process failed", "error", err)
} }
}(p) }(p)
@ -62,8 +64,8 @@ func startPluginAndKeepItAlive(ctx context.Context, p *plugins.Plugin) error {
} }
// keepPluginAlive will restart the plugin if the process is killed or exits // keepPluginAlive will restart the plugin if the process is killed or exits
func keepPluginAlive(p *plugins.Plugin) error { func (s *Service) keepPluginAlive(p *plugins.Plugin) error {
ticker := time.NewTicker(keepPluginAliveTickerDuration) ticker := time.NewTicker(s.keepPluginAliveTickerDuration)
for { for {
<-ticker.C <-ticker.C

@ -4,7 +4,6 @@ import (
"context" "context"
"sync" "sync"
"testing" "testing"
"time"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
@ -15,7 +14,11 @@ import (
) )
func TestProcessManager_Start(t *testing.T) { func TestProcessManager_Start(t *testing.T) {
t.Parallel()
t.Run("Plugin state determines process start", func(t *testing.T) { t.Run("Plugin state determines process start", func(t *testing.T) {
t.Parallel()
tcs := []struct { tcs := []struct {
name string name string
managed bool managed bool
@ -52,14 +55,20 @@ func TestProcessManager_Start(t *testing.T) {
}, },
} }
for _, tc := range tcs { for _, tc := range tcs {
// create a local copy of "tc" to allow concurrent access within tests to the different items of testCases,
// otherwise it would be like a moving pointer while tests run in parallel
tc := tc
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
t.Parallel()
bp := fakes.NewFakeBackendPlugin(tc.managed) bp := fakes.NewFakeBackendPlugin(tc.managed)
p := createPlugin(t, bp, func(plugin *plugins.Plugin) { p := createPlugin(t, bp, func(plugin *plugins.Plugin) {
plugin.Backend = tc.backend plugin.Backend = tc.backend
plugin.SignatureError = tc.signatureError plugin.SignatureError = tc.signatureError
}) })
m := &Service{} m := ProvideService()
err := m.Start(context.Background(), p) err := m.Start(context.Background(), p)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, tc.expectedStartCount, bp.StartCount) require.Equal(t, tc.expectedStartCount, bp.StartCount)
@ -74,18 +83,15 @@ func TestProcessManager_Start(t *testing.T) {
}) })
t.Run("Won't stop the plugin if the context is cancelled", func(t *testing.T) { t.Run("Won't stop the plugin if the context is cancelled", func(t *testing.T) {
t.Parallel()
bp := fakes.NewFakeBackendPlugin(true) bp := fakes.NewFakeBackendPlugin(true)
p := createPlugin(t, bp, func(plugin *plugins.Plugin) { p := createPlugin(t, bp, func(plugin *plugins.Plugin) {
plugin.Backend = true plugin.Backend = true
}) })
tickerDuration := keepPluginAliveTickerDuration m := ProvideService()
keepPluginAliveTickerDuration = 1 * time.Millisecond m.keepPluginAliveTickerDuration = 1
defer func() {
keepPluginAliveTickerDuration = tickerDuration
}()
m := &Service{}
ctx := context.Background() ctx := context.Background()
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
err := m.Start(ctx, p) err := m.Start(ctx, p)
@ -100,7 +106,11 @@ func TestProcessManager_Start(t *testing.T) {
} }
func TestProcessManager_Stop(t *testing.T) { func TestProcessManager_Stop(t *testing.T) {
t.Parallel()
t.Run("Can stop a running plugin", func(t *testing.T) { t.Run("Can stop a running plugin", func(t *testing.T) {
t.Parallel()
pluginID := "test-datasource" pluginID := "test-datasource"
bp := fakes.NewFakeBackendPlugin(true) bp := fakes.NewFakeBackendPlugin(true)
@ -109,7 +119,7 @@ func TestProcessManager_Stop(t *testing.T) {
plugin.Backend = true plugin.Backend = true
}) })
m := &Service{} m := ProvideService()
err := m.Stop(context.Background(), p) err := m.Stop(context.Background(), p)
require.NoError(t, err) require.NoError(t, err)
@ -120,18 +130,21 @@ func TestProcessManager_Stop(t *testing.T) {
} }
func TestProcessManager_ManagedBackendPluginLifecycle(t *testing.T) { func TestProcessManager_ManagedBackendPluginLifecycle(t *testing.T) {
bp := fakes.NewFakeBackendPlugin(true) t.Parallel()
p := createPlugin(t, bp, func(plugin *plugins.Plugin) {
plugin.Backend = true
})
m := &Service{} t.Run("When plugin process is killed, the process is restarted", func(t *testing.T) {
t.Parallel()
bp := fakes.NewFakeBackendPlugin(true)
p := createPlugin(t, bp, func(plugin *plugins.Plugin) {
plugin.Backend = true
})
err := m.Start(context.Background(), p) m := ProvideService()
require.NoError(t, err)
require.Equal(t, 1, bp.StartCount) err := m.Start(context.Background(), p)
require.NoError(t, err)
require.Equal(t, 1, bp.StartCount)
t.Run("When plugin process is killed, the process is restarted", func(t *testing.T) {
var wgKill sync.WaitGroup var wgKill sync.WaitGroup
wgKill.Add(1) wgKill.Add(1)
go func() { go func() {

Loading…
Cancel
Save