Add support for delta bundles

Earlier a snapshot bundle would describe the full state of OPA's
policy/data and any update would require first erasing the state from
the existing bundle and then activating the new bundle.

This commit introduces a new bundle type called "delta".
Delta bundles contain patches to data instead of snapshots.
They allow users to efficiently make updates to OPA's data
cache.

Signed-off-by: Ashutosh Narkar <anarkar4387@gmail.com>

download/config.go

@@ -14,6 +14,12 @@ import (
 const (
 	defaultMinDelaySeconds = int64(60)
 	defaultMaxDelaySeconds = int64(120)
+
+	// deltaBundleMode indicates that OPA supports delta bundle processing
+	deltaBundleMode = "delta"
+
+	// defaultBundleMode indicates that OPA supports snapshot bundle processing
+	defaultBundleMode = "snapshot"
 )
 
 // PollingConfig represents polling configuration for the downloader.

download/download.go

@@ -14,6 +14,7 @@ import (
 	"net/http"
 	"path"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -260,8 +261,12 @@ func (d *Downloader) download(ctx context.Context, m metrics.Metrics) (*download
 	d.logger.Debug("Download starting.")
 
 	d.client = d.client.WithHeader("If-None-Match", d.etag)
+
+	preferences := []string{fmt.Sprintf("modes=%v,%v", defaultBundleMode, deltaBundleMode)}
+
 	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)))
+		wait := fmt.Sprintf("wait=%s", strconv.FormatInt(*d.config.Polling.LongPollingTimeoutSeconds, 10))
+		preferences = append(preferences, wait)
 
 		// fetch existing response header timeout value on the http client's transport and
 		// clear it for the long poll request
@@ -271,10 +276,11 @@ 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")
 	}
 
+	preferValue := fmt.Sprintf("%v", strings.Join(preferences, ";"))
+	d.client = d.client.WithHeader("Prefer", preferValue)
+
 	m.Timer(metrics.BundleRequest).Start()
 	resp, err := d.client.Do(ctx, "GET", d.path)
 	m.Timer(metrics.BundleRequest).Stop()

download/download_test.go

@@ -153,6 +153,37 @@ func TestStopWithMultipleCalls(t *testing.T) {
 	}
 }
 
+func TestStartStopWithDeltaBundleMode(t *testing.T) {
+	ctx := context.Background()
+
+	updates := make(chan *Update)
+
+	config := Config{}
+
+	if err := config.ValidateAndInjectDefaults(); err != nil {
+		t.Fatal(err)
+	}
+
+	fixture := newTestFixture(t)
+
+	d := New(config, fixture.client, "/bundles/test/bundle2").WithCallback(func(_ context.Context, u Update) {
+		updates <- &u
+	})
+
+	d.Start(ctx)
+
+	// Give time for some download events to occur
+	time.Sleep(1 * time.Second)
+
+	u1 := <-updates
+
+	if u1.Bundle == nil || u1.Bundle.Manifest.Revision != deltaBundleMode {
+		t.Fatal("expected delta bundle but got:", u1)
+	}
+
+	d.Stop(ctx)
+}
+
 func TestStartStopWithLongPollNotSupported(t *testing.T) {
 	ctx := context.Background()
 
@@ -584,7 +615,8 @@ func TestOneShotLongPollingSwitch(t *testing.T) {
 func TestOneShotNotLongPollingSwitch(t *testing.T) {
 	ctx := context.Background()
 	config := Config{}
-	config.Polling.LongPollingTimeoutSeconds = nil
+	timeout := int64(3)
+	config.Polling.LongPollingTimeoutSeconds = &timeout
 	if err := config.ValidateAndInjectDefaults(); err != nil {
 		t.Fatal(err)
 	}
@@ -599,7 +631,7 @@ func TestOneShotNotLongPollingSwitch(t *testing.T) {
 	if err != nil {
 		t.Fatal("Unexpected:", err)
 	}
-	if fixture.d.longPollingEnabled != true {
+	if !fixture.d.longPollingEnabled {
 		t.Fatal("Expected long polling to be enabled")
 	}
 
@@ -624,6 +656,12 @@ type testFixture struct {
 
 func newTestFixture(t *testing.T) testFixture {
 
+	patch := bundle.PatchOperation{
+		Op:    "upsert",
+		Path:  "/a/c/d",
+		Value: []string{"foo", "bar"},
+	}
+
 	ts := testServer{
 		t:       t,
 		expAuth: "Bearer secret",
@@ -645,6 +683,12 @@ func newTestFixture(t *testing.T) testFixture {
 					},
 				},
 			},
+			"test/bundle2": {
+				Manifest: bundle.Manifest{
+					Revision: deltaBundleMode,
+				},
+				Patch: bundle.Patch{Data: []bundle.PatchOperation{patch}},
+			},
 		},
 	}
 
@@ -709,12 +753,8 @@ type testServer struct {
 func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
 
 	if t.longPoll {
-		parts := strings.Split(r.Header.Get("Prefer"), "=")
-		if len(parts) != 2 {
-			panic("Invalid \"wait\" Preference")
-		}
-
-		timeout, err := strconv.Atoi(parts[1])
+		wait := getPreferHeaderField(r, "wait")
+		timeout, err := strconv.Atoi(wait)
 		if err != nil {
 			panic(err)
 		}
@@ -742,6 +782,23 @@ func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// check to verify if server can send a delta bundle to OPA
+	if b.Manifest.Revision == deltaBundleMode {
+		modes := strings.Split(getPreferHeaderField(r, "modes"), ",")
+
+		found := false
+		for _, m := range modes {
+			if m == deltaBundleMode {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			panic("delta bundle requested but OPA does not support it")
+		}
+	}
+
 	contentTypeShouldBeSend := true
 	if t.expEtag != "" {
 		etag := r.Header.Get("If-None-Match")
@@ -786,3 +843,17 @@ func (t *testServer) start() {
 func (t *testServer) stop() {
 	t.server.Close()
 }
+
+func getPreferHeaderField(r *http.Request, field string) string {
+	for _, line := range r.Header.Values("prefer") {
+		for _, part := range strings.Split(line, ";") {
+			preference := strings.Split(strings.TrimSpace(part), "=")
+			if len(preference) == 2 {
+				if strings.ToLower(preference[0]) == field {
+					return preference[1]
+				}
+			}
+		}
+	}
+	return ""
+}