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

download: fix unintended switch between long/regular polling on 304 HTTP status (#3926)

Browse Source 
Previously a 304 HTTP status code would make the downloader switch
from long polling to regular polling. This is not desired, as it's in line with
the HTTP spec not to have a Content-Type header for 304 responses.
The absent Content-Type header had triggered the switch back to regular
polling.

Fixes: #3923

Signed-off-by: Gasc Florian <florian.gasc@gmail.com>
This commit is contained in:
floriangasc
2021-12-10 12:20:44 +01:00
committed by GitHub
parent 30a8c2fcab
commit bbe8afda6e
2 changed files with 159 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
download/download.go
+34 -37
View File
@@ -48,21 +48,22 @@ type Update struct {
// updates from the remote HTTP endpoint that the client is configured to
// connect to.
type Downloader struct {
config Config // downloader configuration for tuning polling and other downloader behaviour
client rest.Client // HTTP client to use for bundle downloading
path string // path to use in bundle download request
trigger chan chan struct{} // channel to signal downloads when manual triggering is enabled
stop chan chan struct{} // used to signal plugin to stop running
f func(context.Context, Update) // callback function invoked when download updates occur
etag string // HTTP Etag for caching purposes
sizeLimitBytes *int64 // max bundle file size in bytes (passed to reader)
bvc *bundle.VerificationConfig
respHdrTimeoutSec int64
wg sync.WaitGroup
logger logging.Logger
mtx sync.Mutex
stopped bool
persist bool
config Config // downloader configuration for tuning polling and other downloader behaviour
client rest.Client // HTTP client to use for bundle downloading
path string // path to use in bundle download request
trigger chan chan struct{} // channel to signal downloads when manual triggering is enabled
stop chan chan struct{} // used to signal plugin to stop running
f func(context.Context, Update) // callback function invoked when download updates occur
etag string // HTTP Etag for caching purposes
sizeLimitBytes *int64 // max bundle file size in bytes (passed to reader)
bvc *bundle.VerificationConfig
respHdrTimeoutSec int64
wg sync.WaitGroup
logger logging.Logger
mtx sync.Mutex
stopped bool
persist bool
longPollingEnabled bool
}
type downloaderResponse struct {
@@ -75,12 +76,13 @@ type downloaderResponse struct {
// New returns a new Downloader that can be started.
func New(config Config, client rest.Client, path string) *Downloader {
return &Downloader{
config: config,
client: client,
path: path,
trigger: make(chan chan struct{}),
stop: make(chan chan struct{}),
logger: client.Logger(),
config: config,
client: client,
path: path,
trigger: make(chan chan struct{}),
stop: make(chan chan struct{}),
logger: client.Logger(),
longPollingEnabled: config.Polling.LongPollingTimeoutSeconds != nil,
}
}
@@ -131,7 +133,7 @@ func (d *Downloader) Trigger(ctx context.Context) error {
done := make(chan error)
go func() {
_, err := d.oneShot(ctx)
err := d.oneShot(ctx)
if err != nil {
d.logger.Error("Bundle download failed: %v.", err)
if ctx.Err() == nil {
@@ -197,7 +199,7 @@ func (d *Downloader) loop(ctx context.Context) {
var delay time.Duration
longPoll, err := d.oneShot(ctx)
err := d.oneShot(ctx)
if ctx.Err() != nil {
return
@@ -206,16 +208,11 @@ func (d *Downloader) loop(ctx context.Context) {
if err != nil {
delay = util.DefaultBackoff(float64(minRetryDelay), float64(*d.config.Polling.MaxDelaySeconds), retry)
} else {
if !longPoll {
if d.config.Polling.LongPollingTimeoutSeconds != nil {
d.config.Polling.LongPollingTimeoutSeconds = nil
}
if !d.longPollingEnabled {
// revert the response header timeout value on the http client's transport
if *d.client.Config().ResponseHeaderTimeoutSeconds == 0 {
d.client = d.client.SetResponseHeaderTimeout(&d.respHdrTimeoutSec)
}
min := float64(*d.config.Polling.MinDelaySeconds)
max := float64(*d.config.Polling.MaxDelaySeconds)
delay = time.Duration(((max - min) * rand.Float64()) + min)
@@ -237,7 +234,7 @@ func (d *Downloader) loop(ctx context.Context) {
}
}
func (d *Downloader) oneShot(ctx context.Context) (bool, error) {
func (d *Downloader) oneShot(ctx context.Context) error {
m := metrics.New()
resp, err := d.download(ctx, m)
@@ -247,25 +244,23 @@ func (d *Downloader) oneShot(ctx context.Context) (bool, error) {
if d.f != nil {
d.f(ctx, Update{ETag: "", Bundle: nil, Error: err, Metrics: m, Raw: nil})
}
return false, err
return err
}
d.etag = resp.etag
d.longPollingEnabled = resp.longPoll
if d.f != nil {
d.f(ctx, Update{ETag: resp.etag, Bundle: resp.b, Error: nil, Metrics: m, Raw: resp.raw})
}
return resp.longPoll, nil
return nil
}
func (d *Downloader) download(ctx context.Context, m metrics.Metrics) (*downloaderResponse, error) {
d.logger.Debug("Download starting.")
d.client = d.client.WithHeader("If-None-Match", d.etag)
if d.config.Polling.LongPollingTimeoutSeconds != nil {
if d.longPollingEnabled && d.config.Polling.LongPollingTimeoutSeconds != nil {
d.client = d.client.WithHeader("Prefer", fmt.Sprintf("wait=%s", strconv.FormatInt(*d.config.Polling.LongPollingTimeoutSeconds, 10)))
// fetch existing response header timeout value on the http client's transport and
@@ -276,6 +271,8 @@ func (d *Downloader) download(ctx context.Context, m metrics.Metrics) (*download
t := int64(0)
d.client = d.client.SetResponseHeaderTimeout(&t)
}
} else {
d.client = d.client.WithHeader("Prefer", "wait=0")
}
m.Timer(metrics.BundleRequest).Start()
@@ -337,7 +334,7 @@ func (d *Downloader) download(ctx context.Context, m metrics.Metrics) (*download
b: nil,
raw: nil,
etag: etag,
longPoll: isLongPollSupported(resp.Header),
longPoll: d.longPollingEnabled,
}, nil
case http.StatusNotFound:
return nil, fmt.Errorf("server replied with not found")
download/download_test.go
+125 -25
View File
@@ -2,6 +2,7 @@
// Use of this source code is governed by an Apache2
// license that can be found in the LICENSE file.
//go:build slow
// +build slow
package download
@@ -19,6 +20,7 @@ import (
"testing"
"time"
"github.com/open-policy-agent/opa/metrics"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/bundle"
@@ -258,7 +260,7 @@ func TestEtagCachingLifecycle(t *testing.T) {
// simulate downloader error on first bundle download
fixture.server.expCode = 500
fixture.server.expEtag = "some etag value"
_, err := fixture.d.oneShot(ctx)
err := fixture.d.oneShot(ctx)
if err == nil {
t.Fatal("Expected error but got nil")
} else if len(fixture.updates) != 1 {
@@ -269,7 +271,7 @@ func TestEtagCachingLifecycle(t *testing.T) {
// simulate successful bundle activation and check updated etag on the downloader
fixture.server.expCode = 0
_, err = fixture.d.oneShot(ctx)
err = fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if len(fixture.updates) != 2 {
@@ -280,7 +282,7 @@ func TestEtagCachingLifecycle(t *testing.T) {
// simulate another successful bundle activation and check updated etag on the downloader
fixture.server.expEtag = "some etag value - 2"
_, err = fixture.d.oneShot(ctx)
err = fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if len(fixture.updates) != 3 {
@@ -292,7 +294,7 @@ func TestEtagCachingLifecycle(t *testing.T) {
// simulate bundle activation error and check etag is set from the last successful activation
fixture.mockBundleActivationError = true
fixture.server.expEtag = "some newer etag value - 3"
_, err = fixture.d.oneShot(ctx)
err = fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if len(fixture.updates) != 4 {
@@ -304,7 +306,7 @@ func TestEtagCachingLifecycle(t *testing.T) {
// simulate successful bundle activation and check updated etag on the downloader
fixture.server.expCode = 0
fixture.mockBundleActivationError = false
_, err = fixture.d.oneShot(ctx)
err = fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if len(fixture.updates) != 5 {
@@ -315,7 +317,7 @@ func TestEtagCachingLifecycle(t *testing.T) {
// simulate downloader error and check etag is set from the last successful activation
fixture.server.expCode = 500
_, err = fixture.d.oneShot(ctx)
err = fixture.d.oneShot(ctx)
if err == nil {
t.Fatal("Expected error but got nil")
} else if len(fixture.updates) != 6 {
@@ -327,7 +329,7 @@ func TestEtagCachingLifecycle(t *testing.T) {
// simulate bundle activation error and check etag is set from the last successful activation
fixture.mockBundleActivationError = true
fixture.server.expCode = 0
_, err = fixture.d.oneShot(ctx)
err = fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if len(fixture.updates) != 7 {
@@ -346,7 +348,7 @@ func TestFailureAuthn(t *testing.T) {
d := New(Config{}, fixture.client, "/bundles/test/bundle1")
_, err := d.oneShot(ctx)
err := d.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
}
@@ -361,7 +363,7 @@ func TestFailureNotFound(t *testing.T) {
d := New(Config{}, fixture.client, "/bundles/test/non-existent")
_, err := d.oneShot(ctx)
err := d.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
}
@@ -376,7 +378,7 @@ func TestFailureUnexpected(t *testing.T) {
d := New(Config{}, fixture.client, "/bundles/test/bundle1")
_, err := d.oneShot(ctx)
err := d.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
}
@@ -395,7 +397,7 @@ func TestEtagInResponse(t *testing.T) {
fixture.server.expEtag = "some etag value"
_, err := fixture.d.oneShot(ctx)
err := fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if len(fixture.updates) != 1 {
@@ -409,7 +411,7 @@ func TestEtagInResponse(t *testing.T) {
t.Errorf("Expected bundle in response")
}
_, err = fixture.d.oneShot(ctx)
err = fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if len(fixture.updates) != 2 {
@@ -518,6 +520,99 @@ func TestTriggerManualWithTimeout(t *testing.T) {
d.Stop(ctx)
}
func TestDownloadLongPollNotModifiedOn304(t *testing.T) {
ctx := context.Background()
config := Config{}
timeout := int64(3) // this will result in the test server sleeping for 3 seconds
config.Polling.LongPollingTimeoutSeconds = &timeout
if err := config.ValidateAndInjectDefaults(); err != nil {
t.Fatal(err)
}
fixture := newTestFixture(t)
fixture.d = New(config, fixture.client, "/bundles/test/bundle1").WithCallback(fixture.oneShot)
fixture.server.longPoll = true
fixture.server.expEtag = "foo"
fixture.d.etag = fixture.server.expEtag // not modified
fixture.server.expCode = 0
defer fixture.server.stop()
resp, err := fixture.d.download(ctx, metrics.New())
if err != nil {
t.Fatal("Unexpected:", err)
}
if resp.longPoll != fixture.d.longPollingEnabled {
t.Fatalf("Expected same value for longPoll and longPollingEnabled")
}
}
func TestOneShotLongPollingSwitch(t *testing.T) {
ctx := context.Background()
config := Config{}
timeout := int64(3) // this will result in the test server sleeping for 3 seconds
config.Polling.LongPollingTimeoutSeconds = &timeout
if err := config.ValidateAndInjectDefaults(); err != nil {
t.Fatal(err)
}
fixture := newTestFixture(t)
fixture.d = New(config, fixture.client, "/bundles/test/bundle1").WithCallback(fixture.oneShot)
fixture.server.expCode = 0
defer fixture.server.stop()
fixture.server.longPoll = true
err := fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
}
if fixture.d.longPollingEnabled != fixture.server.longPoll {
t.Fatalf("Expected same value for longPoll and longPollingEnabled")
}
fixture.server.longPoll = false
err = fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
}
if fixture.d.longPollingEnabled != fixture.server.longPoll {
t.Fatalf("Expected same value for longPollingEnabled and longPoll")
}
}
func TestOneShotNotLongPollingSwitch(t *testing.T) {
ctx := context.Background()
config := Config{}
config.Polling.LongPollingTimeoutSeconds = nil
if err := config.ValidateAndInjectDefaults(); err != nil {
t.Fatal(err)
}
fixture := newTestFixture(t)
fixture.d = New(config, fixture.client, "/bundles/test/bundle1").WithCallback(fixture.oneShot)
fixture.server.expCode = 0
defer fixture.server.stop()
fixture.server.longPoll = true
err := fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
}
if fixture.d.longPollingEnabled != true {
t.Fatal("Expected long polling to be enabled")
}
fixture.server.longPoll = false
err = fixture.d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
}
if fixture.d.longPollingEnabled {
t.Fatal("Expected long polling to be disabled")
}
}
type testFixture struct {
d *Downloader
client rest.Client
@@ -601,14 +696,15 @@ func (t *testFixture) oneShot(ctx context.Context, u Update) {
}
type testServer struct {
t *testing.T
expCode int
expEtag string
expAuth string
bundles map[string]bundle.Bundle
server *httptest.Server
etagInResponse bool
longPoll bool
t *testing.T
expCode int
expEtag string
expAuth string
bundles map[string]bundle.Bundle
server *httptest.Server
etagInResponse bool
longPoll bool
opaVendorMediaTypeEnabled bool
}
func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
@@ -624,13 +720,8 @@ func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
panic(err)
}
// indicate server supports long polling
w.Header().Add("Content-Type", "application/vnd.openpolicyagent.bundles")
// simulate long operation
time.Sleep(time.Duration(timeout) * time.Second)
} else {
w.Header().Add("Content-Type", "application/gzip")
}
if t.expCode != 0 {
@@ -652,9 +743,11 @@ func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
return
}
contentTypeShouldBeSend := true
if t.expEtag != "" {
etag := r.Header.Get("If-None-Match")
if etag == t.expEtag {
contentTypeShouldBeSend = false
if t.etagInResponse {
w.Header().Add("Etag", t.expEtag)
}
@@ -663,6 +756,13 @@ func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
}
}
if t.longPoll && contentTypeShouldBeSend {
// in 304 Content-Type is not send according https://datatracker.ietf.org/doc/html/rfc7232#section-4.1
w.Header().Add("Content-Type", "application/vnd.openpolicyagent.bundles")
} else {
w.Header().Add("Content-Type", "application/gzip")
}
if t.expEtag != "" {
w.Header().Add("Etag", t.expEtag)
}