ewilliams28/releases
1
0
Fork 0
mirror of https://github.com/open-policy-agent/opa.git synced 2026-08-12 19:32:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

plugins: support graceful shutdown through SDK (#4119)

Browse Source 
With this change, the manager will respect the shutdown period if
it was supplied, otherwise it will use the passed context. This way,
SDK users can rely on the context (because the SDK doesn't set
the graceful shutdown period), but other callers are unaffected.

The added test is in the SDK, because that's where the problem had
manifested (#3980): when calling Stop(ctx) through the SDK, the
plugins.Manager's Stop function had set a smaller timeout (0, due
to the structs default value that hadn't been set to anything else),
and that would effectively leave the plugins zero time to cleanup.

Fixes #3980.

Signed-off-by: Stephan Renatus <stephan.renatus@gmail.com>
This commit is contained in:
Stephan Renatus
2021-12-14 09:43:22 +01:00
committed by GitHub
parent 16919a55b2
commit 883dc8817f
2 changed files with 105 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/plugins.go
+11 -3
View File
@@ -588,8 +588,11 @@ func (m *Manager) Start(ctx context.Context) error {
return nil
}
// Stop stops the manager, stopping all the plugins registered with it. Any plugin that needs to perform cleanup should
// do so within the duration of the graceful shutdown period passed with the context as a timeout.
// Stop stops the manager, stopping all the plugins registered with it.
// Any plugin that needs to perform cleanup should do so within the duration
// of the graceful shutdown period passed with the context as a timeout.
// Note that a graceful shutdown period configured with the Manager instance
// will override the timeout of the passed in context (if applicable).
func (m *Manager) Stop(ctx context.Context) {
var toStop []Plugin
@@ -602,7 +605,12 @@ func (m *Manager) Stop(ctx context.Context) {
}
}()
ctx, cancel := context.WithTimeout(ctx, time.Duration(m.gracefulShutdownPeriod)*time.Second)
var cancel context.CancelFunc
if m.gracefulShutdownPeriod > 0 {
ctx, cancel = context.WithTimeout(ctx, time.Duration(m.gracefulShutdownPeriod)*time.Second)
} else {
ctx, cancel = context.WithCancel(ctx)
}
defer cancel()
for i := range toStop {
toStop[i].Stop(ctx)
sdk/opa_test.go
+94 -54
View File
@@ -5,12 +5,12 @@
package sdk_test
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"reflect"
"strings"
"testing"
"time"
@@ -22,31 +22,39 @@ import (
"github.com/open-policy-agent/opa/version"
)
// Plugin creates an empty plugin to test plugin initialization
// Plugin creates an empty plugin to test plugin initialization and shutdown
type plugin struct {
manager *plugins.Manager
manager *plugins.Manager
shutdown time.Duration // to simulate a shutdown that takes some time
}
type factory struct{}
type factory struct {
shutdown time.Duration
}
func (p *plugin) Start(ctx context.Context) error {
func (p *plugin) Start(context.Context) error {
p.manager.UpdatePluginStatus("test_plugin", &plugins.Status{State: plugins.StateOK})
return nil
}
func (p *plugin) Stop(ctx context.Context) {
}
func (p *plugin) Reconfigure(ctx context.Context, config interface{}) {
}
func (factory) New(manager *plugins.Manager, config interface{}) plugins.Plugin {
return &plugin{
manager: manager,
select {
case <-ctx.Done():
case <-time.After(p.shutdown):
}
}
func (factory) Validate(manager *plugins.Manager, config []byte) (interface{}, error) {
func (*plugin) Reconfigure(context.Context, interface{}) {
}
func (f factory) New(manager *plugins.Manager, config interface{}) plugins.Plugin {
return &plugin{
manager: manager,
shutdown: f.shutdown,
}
}
func (factory) Validate(*plugins.Manager, []byte) (interface{}, error) {
return nil, nil
}
@@ -54,14 +62,14 @@ func TestPlugins(t *testing.T) {
ctx := context.Background()
config := []byte(`{
config := `{
"plugins": {
"test_plugin": {}
}
}`)
}`
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
Plugins: map[string]plugins.Factory{
"test_plugin": factory{},
},
@@ -94,7 +102,7 @@ func TestDecision(t *testing.T) {
defer server.Stop()
config := []byte(fmt.Sprintf(`{
config := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -105,10 +113,10 @@ func TestDecision(t *testing.T) {
"resource": "/bundles/bundle.tar.gz"
}
}
}`, server.URL()))
}`, server.URL())
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
})
if err != nil {
t.Fatal(err)
@@ -149,7 +157,7 @@ func TestUndefinedError(t *testing.T) {
defer server.Stop()
config := []byte(fmt.Sprintf(`{
config := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -160,10 +168,10 @@ func TestUndefinedError(t *testing.T) {
"resource": "/bundles/bundle.tar.gz"
}
}
}`, server.URL()))
}`, server.URL())
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
})
if err != nil {
t.Fatal(err)
@@ -194,7 +202,7 @@ func TestDecisionLogging(t *testing.T) {
defer server.Stop()
config := []byte(fmt.Sprintf(`{
config := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -208,11 +216,11 @@ func TestDecisionLogging(t *testing.T) {
"decision_logs": {
"console": true
}
}`, server.URL()))
}`, server.URL())
testLogger := loggingtest.New()
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
ConsoleLogger: testLogger,
})
if err != nil {
@@ -256,7 +264,7 @@ func TestQueryCaching(t *testing.T) {
defer server.Stop()
config := []byte(fmt.Sprintf(`{
config := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -270,11 +278,11 @@ func TestQueryCaching(t *testing.T) {
"decision_logs": {
"console": true
}
}`, server.URL()))
}`, server.URL())
testLogger := loggingtest.New()
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
ConsoleLogger: testLogger,
})
if err != nil {
@@ -331,7 +339,7 @@ func TestDiscovery(t *testing.T) {
defer server.Stop()
config := []byte(fmt.Sprintf(`{
config := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -340,10 +348,10 @@ func TestDiscovery(t *testing.T) {
"discovery": {
"resource": "/bundles/discovery.tar.gz"
}
}`, server.URL()))
}`, server.URL())
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
})
if err != nil {
t.Fatal(err)
@@ -377,7 +385,7 @@ func TestAsync(t *testing.T) {
defer server.Stop()
config := []byte(fmt.Sprintf(`{
config := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -388,10 +396,10 @@ func TestAsync(t *testing.T) {
"resource": "/bundles/bundle.tar.gz"
}
}
}`, server.URL()))
}`, server.URL())
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
Ready: readyCh,
})
if err != nil {
@@ -422,7 +430,7 @@ func TestCancelStartup(t *testing.T) {
server := sdktest.MustNewServer()
defer server.Stop()
config := []byte(fmt.Sprintf(`{
config := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -433,20 +441,52 @@ func TestCancelStartup(t *testing.T) {
"resource": "/bundles/doesnotexist.tar.gz"
}
}
}`, server.URL()))
}`, server.URL())
// Server will return 404 responses because bundle does not exist. OPA should timeout.
ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*100)
defer cancel()
_, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
})
if !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("expected deadline exceeded error but got %v", err)
}
}
// TestStopWithDeadline asserts that a graceful shutdown of the SDK is possible.
func TestStopWithDeadline(t *testing.T) {
ctx := context.Background()
opa, err := sdk.New(ctx, sdk.Options{
Config: strings.NewReader(`{
"plugins": {
"test_plugin": {}
}
}`),
Plugins: map[string]plugins.Factory{
"test_plugin": factory{shutdown: time.Second},
},
})
if err != nil {
t.Fatal(err)
}
const timeout = 20 * time.Millisecond
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
before := time.Now()
opa.Stop(ctx) // 1m timeout is ignored
dur := time.Since(before)
diff := dur - timeout
maxDelta := 6 * time.Millisecond
if diff > maxDelta || diff < -maxDelta {
t.Errorf("expected shutdown to have %v grace period, measured shutdown in %v (max delta %v)", timeout, dur, maxDelta)
}
}
func TestConfigAsYAML(t *testing.T) {
server := sdktest.MustNewServer(
@@ -456,16 +496,16 @@ func TestConfigAsYAML(t *testing.T) {
)
defer server.Stop()
config := []byte(fmt.Sprintf(`services:
config := fmt.Sprintf(`services:
test:
url: %q
bundles:
test:
resource: "/bundles/bundle.tar.gz"`, server.URL()))
resource: "/bundles/bundle.tar.gz"`, server.URL())
ctx := context.Background()
_, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
})
if err != nil {
t.Fatal(err)
@@ -486,7 +526,7 @@ func TestConfigure(t *testing.T) {
defer server.Stop()
// Startup new OPA with first config.
config1 := []byte(fmt.Sprintf(`{
config1 := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -497,11 +537,11 @@ func TestConfigure(t *testing.T) {
"resource": "/bundles/bundle1.tar.gz"
}
}
}`, server.URL()))
}`, server.URL())
ctx := context.Background()
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config1),
Config: strings.NewReader(config1),
})
if err != nil {
t.Fatal(err)
@@ -518,7 +558,7 @@ func TestConfigure(t *testing.T) {
}
// Reconfigure with new config to make sure update is picked up.
config2 := []byte(fmt.Sprintf(`{
config2 := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -529,10 +569,10 @@ func TestConfigure(t *testing.T) {
"resource": "/bundles/bundle2.tar.gz"
}
}
}`, server.URL()))
}`, server.URL())
err = opa.Configure(ctx, sdk.ConfigOptions{
Config: bytes.NewReader(config2),
Config: strings.NewReader(config2),
})
if err != nil {
t.Fatal(err)
@@ -549,7 +589,7 @@ func TestConfigure(t *testing.T) {
// Reconfigure w/ same config to verify that readiness channel is closed.
ch := make(chan struct{})
err = opa.Configure(ctx, sdk.ConfigOptions{
Config: bytes.NewReader(config2),
Config: strings.NewReader(config2),
Ready: ch,
})
if err != nil {
@@ -573,7 +613,7 @@ func TestOpaVersion(t *testing.T) {
defer server.Stop()
config := []byte(fmt.Sprintf(`{
config := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -584,10 +624,10 @@ func TestOpaVersion(t *testing.T) {
"resource": "/bundles/bundle.tar.gz"
}
}
}`, server.URL()))
}`, server.URL())
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
})
if err != nil {
t.Fatal(err)
@@ -627,7 +667,7 @@ func TestOpaRuntimeConfig(t *testing.T) {
testBundleResource := "/bundles/bundle.tar.gz"
testLabel := "a label"
config := []byte(fmt.Sprintf(`{
config := fmt.Sprintf(`{
"services": {
"test": {
"url": %q
@@ -641,10 +681,10 @@ func TestOpaRuntimeConfig(t *testing.T) {
"labels": {
"test": %q
}
}`, server.URL(), testBundleResource, testLabel))
}`, server.URL(), testBundleResource, testLabel)
opa, err := sdk.New(ctx, sdk.Options{
Config: bytes.NewReader(config),
Config: strings.NewReader(config),
})
if err != nil {
t.Fatal(err)