This change allows the HTTP clients to consume and send gzip compressed response and request body. (#5696)

It is available for the following REST API endpoints:
- GET & POST HTTP methods on /v0/data & /v1/data endpoints
- POST HTTP method on /v1/compile endpoint

HTTP clients can optionally:
- send 'Accept-Encoding: gzip' header and expect a gzip compressed body and a Content-Encoding: gzip response header. The server will send the content encoded as gzip only after a threshold defined by server.encoding.gzip.min_length (default value is 1024). If the size is below the threshold, the body is not compressed
- send 'Content-Encoding: gzip' header and a gzip compressed body and expect the server to correctly interpret the request

Fixes #5310

Signed-off-by: aarnautu <aarnautu@adobe.com>

server/handlers/compress.go

@@ -0,0 +1,194 @@
+package handlers
+
+import (
+	"compress/gzip"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+)
+
+const (
+	acceptEncodingHeader  = "Accept-Encoding"
+	contentEncodingHeader = "Content-Encoding"
+	contentLengthHeader   = "Content-Length"
+	gzipEncodingValue     = "gzip"
+)
+
+// This handler applies only for data and compile endpoints, for selected HTTP methods
+//
+// If the client asked for a gzip response, this handler will buffer the response and
+// wait until it reached a certain threshold. If the threshold is not hit, the uncompressed response is sent
+//
+// If a gzip response is not asked by the client, it'll send the uncompressed response
+//
+// The threshold and the gzip compression level can be modified from server's configuration
+
+func CompressHandler(handler http.Handler, gzipMinLength int, gzipCompressionLevel int) http.Handler {
+	initGzipPool(gzipCompressionLevel)
+
+	return http.HandlerFunc(func(responseWriter http.ResponseWriter, request *http.Request) {
+		enabledForEndpoint := isDataEndpoint(request) || isCompileEndpoint(request)
+		if !enabledForEndpoint {
+			handler.ServeHTTP(responseWriter, request)
+			return
+		}
+
+		responseWriter.Header().Add("Vary", acceptEncodingHeader)
+
+		if !gzipHeaderDetected(request.Header) {
+			handler.ServeHTTP(responseWriter, request)
+			return
+		}
+
+		crw := &compressResponseWriter{
+			ResponseWriter: responseWriter,
+			headerWritten:  false,
+			minlength:      gzipMinLength,
+		}
+		defer crw.Close()
+		handler.ServeHTTP(crw, request)
+	})
+}
+
+type compressResponseWriter struct {
+	gzipWriter *gzip.Writer
+	http.ResponseWriter
+	buffer        []byte
+	statusCode    int
+	headerWritten bool
+	minlength     int
+}
+
+var gzipPool *sync.Pool
+
+func initGzipPool(compressionLevel int) {
+	if gzipPool == nil {
+		gzipPool = &sync.Pool{
+			New: func() interface{} {
+				writer, _ := gzip.NewWriterLevel(io.Discard, compressionLevel)
+				return writer
+			},
+		}
+	}
+}
+
+func (w *compressResponseWriter) WriteHeader(statusCode int) {
+	// save the status code for later use
+	w.statusCode = statusCode
+}
+
+func (w *compressResponseWriter) Write(bytes []byte) (int, error) {
+	if w.isGzipInitialized() {
+		return w.gzipWriter.Write(bytes)
+	}
+
+	// accumulate the buffer
+	w.buffer = append(w.buffer, bytes...)
+
+	// if the buffer is above threshold, use compression
+	if len(w.buffer) >= w.minlength {
+		err := w.doCompressedResponse()
+		if err != nil {
+			return 0, err
+		}
+		return len(bytes), nil
+	}
+
+	// wait for more data
+	return len(bytes), nil
+}
+
+func (w *compressResponseWriter) Flush() {
+	if w.isGzipInitialized() {
+		w.gzipWriter.Flush()
+		flusher, canFlush := w.ResponseWriter.(http.Flusher)
+		if canFlush {
+			flusher.Flush()
+		}
+	}
+}
+
+func (w *compressResponseWriter) Close() error {
+	if !w.isGzipInitialized() {
+		// gzip didn't handle the response, send it plain
+		err := w.doUncompressedResponse()
+		if err != nil {
+			err = fmt.Errorf("error writing uncompressed data: %v", err.Error())
+		}
+		return err
+	}
+
+	err := w.gzipWriter.Close()
+	defer gzipPool.Put(w.gzipWriter)
+	w.gzipWriter = nil
+	return err
+}
+
+func (w *compressResponseWriter) doCompressedResponse() error {
+	w.ResponseWriter.Header().Set(contentEncodingHeader, gzipEncodingValue)
+	w.Header().Del(contentLengthHeader)
+	w.writeHeader()
+	// there's nothing to write
+	if w.buffer == nil || len(w.buffer) <= 0 {
+		return nil
+	}
+	gzipWriter := gzipPool.Get().(*gzip.Writer)
+	gzipWriter.Reset(w.ResponseWriter)
+	w.gzipWriter = gzipWriter
+	_, err := w.gzipWriter.Write(w.buffer)
+	return err
+}
+
+func (w *compressResponseWriter) doUncompressedResponse() error {
+	w.writeHeader()
+	// there's nothing to write
+	if w.buffer == nil {
+		return nil
+	}
+	_, err := w.ResponseWriter.Write(w.buffer)
+	w.buffer = nil
+	return err
+}
+
+func (w *compressResponseWriter) isGzipInitialized() bool {
+	return w.gzipWriter != nil
+}
+
+func (w *compressResponseWriter) writeHeader() {
+	if !w.headerWritten && w.statusCode != 0 {
+		w.ResponseWriter.WriteHeader(w.statusCode)
+		w.headerWritten = true
+	}
+}
+
+func isDataEndpoint(req *http.Request) bool {
+	isPostOrGetMethod := isPostMethod(req) || isGetMethod(req)
+	isV1rV0 := strings.HasPrefix(req.URL.Path, "/v1/data") || strings.HasPrefix(req.URL.Path, "/v0/data")
+	return isPostOrGetMethod && isV1rV0
+}
+
+func isCompileEndpoint(req *http.Request) bool {
+	return isPostMethod(req) && strings.HasPrefix(req.URL.Path, "/v1/compile")
+}
+
+func isPostMethod(req *http.Request) bool {
+	return req.Method == "POST"
+}
+
+func isGetMethod(req *http.Request) bool {
+	return req.Method == "GET"
+}
+
+func gzipHeaderDetected(header http.Header) bool {
+	a := header.Get("Accept-Encoding")
+	parts := strings.Split(a, ",")
+	for _, part := range parts {
+		part = strings.TrimSpace(part)
+		if part == gzipEncodingValue || strings.HasPrefix(part, gzipEncodingValue+";") {
+			return true
+		}
+	}
+	return false
+}

server/handlers/compress_test.go

@@ -0,0 +1,182 @@
+package handlers
+
+import (
+	"bytes"
+	"compress/gzip"
+	"io"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+)
+
+const (
+	gzipEncoding = "gzip"
+	requestBody  = "Hello World!\n"
+)
+
+var defaultCompressionLevel = gzip.BestCompression
+
+type compressHandlerTestScenario struct {
+	path                       string
+	method                     string
+	acceptEncoding             string
+	gzipMinSize                int
+	expectedCompressedResponse bool
+}
+
+func executeRequest(w *httptest.ResponseRecorder, testScenario compressHandlerTestScenario) {
+	CompressHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, err := io.WriteString(w, requestBody)
+		if err != nil {
+			log.Fatalf("Error writing the request body: %v", err)
+		}
+	}), testScenario.gzipMinSize, defaultCompressionLevel).ServeHTTP(w, &http.Request{
+		URL:    &url.URL{Path: testScenario.path},
+		Method: testScenario.method,
+		Header: http.Header{
+			"Accept-Encoding": []string{testScenario.acceptEncoding},
+		},
+	})
+}
+
+func TestCompressHandlerWithGzipOnInScopeEndpoints(t *testing.T) {
+	tests := map[string]compressHandlerTestScenario{
+		"v0PostDataCompressed": {
+			path:                       "/v0/data",
+			method:                     "POST",
+			acceptEncoding:             gzipEncoding,
+			gzipMinSize:                1,
+			expectedCompressedResponse: true,
+		},
+		"v1PostDataCompressed": {
+			path:                       "/v1/data",
+			method:                     "POST",
+			acceptEncoding:             gzipEncoding,
+			gzipMinSize:                1,
+			expectedCompressedResponse: true,
+		},
+		"v1PostCompileCompressed": {
+			path:                       "/v1/compile",
+			method:                     "POST",
+			acceptEncoding:             gzipEncoding,
+			gzipMinSize:                1,
+			expectedCompressedResponse: true,
+		},
+		"v0PostDataUncompressed": {
+			path:                       "/v0/data",
+			method:                     "POST",
+			acceptEncoding:             gzipEncoding,
+			gzipMinSize:                1024,
+			expectedCompressedResponse: false,
+		},
+		"v1PostDataUncompressed": {
+			path:                       "/v1/data",
+			method:                     "POST",
+			acceptEncoding:             gzipEncoding,
+			gzipMinSize:                1024,
+			expectedCompressedResponse: false,
+		},
+		"v1PostCompileUncompressed": {
+			path:                       "/v1/compile",
+			method:                     "POST",
+			acceptEncoding:             gzipEncoding,
+			gzipMinSize:                1024,
+			expectedCompressedResponse: false,
+		},
+		"v1PostCompileAcceptEncodingAll": {
+			path:                       "/v1/compile",
+			method:                     "POST",
+			acceptEncoding:             "*/*",
+			gzipMinSize:                1024,
+			expectedCompressedResponse: false,
+		},
+	}
+	for name, ts := range tests {
+		w := httptest.NewRecorder()
+		executeRequest(w, ts)
+		if w.Result().Header.Get("Vary") != "Accept-Encoding" {
+			t.Error("missing the Vary header")
+		}
+		contentEncodingValue := w.Result().Header.Get("Content-Encoding")
+		if ts.expectedCompressedResponse {
+			if contentEncodingValue != gzipEncoding {
+				t.Errorf("wrong content encoding, got %q want %q", contentEncodingValue, gzipEncoding)
+			}
+			expectedLength := len(zipString(requestBody))
+			receivedLength := w.Body.Len()
+			if receivedLength != expectedLength {
+				t.Errorf("test: %s wrong len, got %d want %d", name, w.Body.Len(), expectedLength)
+			}
+			receivedBody := unzip(w.Body.Bytes())
+			if receivedBody != requestBody {
+				t.Errorf("test: %s wrong body, got %v, want %v", name, receivedBody, requestBody)
+			}
+		} else {
+			if contentEncodingValue == gzipEncoding {
+				t.Errorf("wrong content encoding, got %q want %q", contentEncodingValue, "")
+			}
+			expectedLength := len(requestBody)
+			receivedLength := w.Body.Len()
+			if receivedLength != expectedLength {
+				t.Errorf("test: %s wrong len, got %d want %d", name, w.Body.Len(), expectedLength)
+			}
+			receivedBody := w.Body.String()
+			if receivedBody != requestBody {
+				t.Errorf("test: %s wrong body, got %v, want %v", name, receivedBody, requestBody)
+			}
+		}
+	}
+}
+
+func TestHandlerOnEndpointsWithoutCompression(t *testing.T) {
+	testScenario := compressHandlerTestScenario{
+		path:           "/metrics",
+		method:         "GET",
+		acceptEncoding: gzipEncoding,
+		gzipMinSize:    1,
+	}
+	w := httptest.NewRecorder()
+	executeRequest(w, testScenario)
+	contentEncodingValue := w.Result().Header.Get("Content-Encoding")
+	if contentEncodingValue != "" {
+		t.Errorf("wrong content encoding, got %q want %q", contentEncodingValue, gzipEncoding)
+	}
+
+	expectedLength := len(requestBody)
+	receivedLength := w.Body.Len()
+	if receivedLength != expectedLength {
+		t.Errorf("wrong len, got %d want %d", w.Body.Len(), expectedLength)
+	}
+}
+
+func zipString(input string) []byte {
+	var b bytes.Buffer
+	gz := gzip.NewWriter(&b)
+	if _, err := gz.Write([]byte(input)); err != nil {
+		log.Fatal(err)
+	}
+	if err := gz.Close(); err != nil {
+		log.Fatal(err)
+	}
+	return b.Bytes()
+}
+
+func unzip(body []byte) string {
+	reader := bytes.NewReader(body)
+	gzReader, err := gzip.NewReader(reader)
+	if err != nil {
+		log.Fatalf("Unexpected gzip error: %v", err)
+	}
+	plainOutput, err := io.ReadAll(gzReader)
+	if err != nil {
+		log.Fatalf("Unexpected gzip error: %v", err)
+	}
+	err = gzReader.Close()
+	if err != nil {
+		log.Fatalf("Unexpected gzip close err: %v", err)
+	}
+	return string(plainOutput)
+}

server/server.go

@@ -6,6 +6,7 @@ package server
 
 import (
 	"bytes"
+	"compress/gzip"
 	"context"
 	"crypto/tls"
 	"crypto/x509"
@@ -24,6 +25,8 @@ import (
 	"sync"
 	"time"
 
+	serverEncodingPlugin "github.com/open-policy-agent/opa/plugins/server/encoding"
+
 	"github.com/gorilla/mux"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
@@ -41,6 +44,7 @@ import (
 	"github.com/open-policy-agent/opa/plugins/status"
 	"github.com/open-policy-agent/opa/rego"
 	"github.com/open-policy-agent/opa/server/authorizer"
+	"github.com/open-policy-agent/opa/server/handlers"
 	"github.com/open-policy-agent/opa/server/identifier"
 	"github.com/open-policy-agent/opa/server/types"
 	"github.com/open-policy-agent/opa/server/writer"
@@ -185,6 +189,11 @@ func (s *Server) Init(ctx context.Context) (*Server, error) {
 
 	// authorizer, if configured, needs the iCache to be set up already
 	s.Handler = s.initHandlerAuth(s.Handler)
+	// compression handler
+	s.Handler, err = s.initHandlerCompression(s.Handler)
+	if err != nil {
+		return nil, err
+	}
 	s.DiagnosticHandler = s.initHandlerAuth(s.DiagnosticHandler)
 
 	return s, s.store.Commit(ctx, txn)
@@ -658,6 +667,21 @@ func (s *Server) initHandlerAuth(handler http.Handler) http.Handler {
 	return handler
 }
 
+func (s *Server) initHandlerCompression(handler http.Handler) (http.Handler, error) {
+	var encodingRawConfig json.RawMessage
+	serverConfig := s.manager.Config.Server
+	if serverConfig != nil {
+		encodingRawConfig = serverConfig.Encoding
+	}
+	encodingConfig, err := serverEncodingPlugin.NewConfigBuilder().WithBytes(encodingRawConfig).Parse()
+	if err != nil {
+		return nil, err
+	}
+	compressHandler := handlers.CompressHandler(handler, *encodingConfig.Gzip.MinLength, *encodingConfig.Gzip.CompressionLevel)
+
+	return compressHandler, nil
+}
+
 func (s *Server) initRouters() {
 	mainRouter := s.router
 	if mainRouter == nil {
@@ -1243,7 +1267,15 @@ func (s *Server) v1CompilePost(w http.ResponseWriter, r *http.Request) {
 	m.Timer(metrics.ServerHandler).Start()
 	m.Timer(metrics.RegoQueryParse).Start()
 
-	request, reqErr := readInputCompilePostV1(r.Body)
+	// decompress the input if sent as zip
+	body, err := readPlainBody(r)
+	if err != nil {
+		reqErr := types.NewErrorV1(types.CodeInvalidParameter, "could not decompress the body")
+		writer.Error(w, http.StatusBadRequest, reqErr)
+		return
+	}
+
+	request, reqErr := readInputCompilePostV1(body)
 	if reqErr != nil {
 		writer.Error(w, http.StatusBadRequest, reqErr)
 		return
@@ -2713,10 +2745,16 @@ func readInputV0(r *http.Request) (ast.Value, error) {
 		return ast.InterfaceToValue(parsed)
 	}
 
+	// decompress the input if sent as zip
+	body, err := readPlainBody(r)
+	if err != nil {
+		return nil, fmt.Errorf("could not decompress the body: %w", err)
+	}
+
 	var x interface{}
 
 	if strings.Contains(r.Header.Get("Content-Type"), "yaml") {
-		bs, err := io.ReadAll(r.Body)
+		bs, err := io.ReadAll(body)
 		if err != nil {
 			return nil, err
 		}
@@ -2726,7 +2764,7 @@ func readInputV0(r *http.Request) (ast.Value, error) {
 			}
 		}
 	} else {
-		dec := util.NewJSONDecoder(r.Body)
+		dec := util.NewJSONDecoder(body)
 		if err := dec.Decode(&x); err != nil && err != io.EOF {
 			return nil, fmt.Errorf("body contains malformed input document: %w", err)
 		}
@@ -2757,11 +2795,17 @@ func readInputPostV1(r *http.Request) (ast.Value, error) {
 
 	var request types.DataRequestV1
 
+	// decompress the input if sent as zip
+	body, err := readPlainBody(r)
+	if err != nil {
+		return nil, fmt.Errorf("could not decompress the body: %w", err)
+	}
+
 	ct := r.Header.Get("Content-Type")
 	// There is no standard for yaml mime-type so we just look for
 	// anything related
 	if strings.Contains(ct, "yaml") {
-		bs, err := io.ReadAll(r.Body)
+		bs, err := io.ReadAll(body)
 		if err != nil {
 			return nil, err
 		}
@@ -2771,7 +2815,7 @@ func readInputPostV1(r *http.Request) (ast.Value, error) {
 			}
 		}
 	} else {
-		dec := util.NewJSONDecoder(r.Body)
+		dec := util.NewJSONDecoder(body)
 		if err := dec.Decode(&request); err != nil && err != io.EOF {
 			return nil, fmt.Errorf("body contains malformed input document: %w", err)
 		}
@@ -2981,3 +3025,19 @@ func annotateSpan(ctx context.Context, decisionID string) {
 	trace.SpanFromContext(ctx).
 		SetAttributes(attribute.String(otelDecisionIDAttr, decisionID))
 }
+
+func readPlainBody(r *http.Request) (io.ReadCloser, error) {
+	if strings.Contains(r.Header.Get("Content-Encoding"), "gzip") {
+		gzReader, err := gzip.NewReader(r.Body)
+		if err != nil {
+			return nil, err
+		}
+		bytesBody, err := io.ReadAll(gzReader)
+		if err != nil {
+			return nil, err
+		}
+		defer gzReader.Close()
+		return io.NopCloser(bytes.NewReader(bytesBody)), err
+	}
+	return r.Body, nil
+}

server/server_test.go

@@ -7,10 +7,13 @@ package server
 
 import (
 	"bytes"
+	"compress/gzip"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"log"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -1763,6 +1766,343 @@ func TestDataPutV1IfNoneMatch(t *testing.T) {
 	}
 }
 
+func TestDataPostV0CompressedResponse(t *testing.T) {
+	tests := []struct {
+		gzipMinLength      int
+		compressedResponse bool
+	}{
+		{
+			gzipMinLength:      3,
+			compressedResponse: true,
+		},
+		{
+			gzipMinLength:      1400,
+			compressedResponse: false,
+		},
+	}
+
+	for _, test := range tests {
+		f := newFixtureWithConfig(t, fmt.Sprintf(`{"server":{"encoding":{"gzip":{"min_length": %d}}}}`, test.gzipMinLength))
+		// create the policy
+		err := f.v1(http.MethodPut, "/policies/test", `package opa.examples
+import input.example.flag
+allow_request { flag == true }
+`, 200, "")
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// execute the request
+		req := newReqV0(http.MethodPost, "/data/opa/examples/allow_request", `{"example": {"flag": true}}`)
+		req.Header.Set("Accept-Encoding", "gzip")
+		f.reset()
+		f.server.Handler.ServeHTTP(f.recorder, req)
+
+		// check for content encoding
+		expectedEncoding := "gzip"
+		if !test.compressedResponse {
+			expectedEncoding = ""
+		}
+		receivedEncodingHeaderValue := f.recorder.Header().Get("Content-Encoding")
+		if receivedEncodingHeaderValue != expectedEncoding {
+			t.Fatalf("Expected Content-Encoding %v but got: %v", expectedEncoding, receivedEncodingHeaderValue)
+		}
+
+		var plainOutput []byte
+		if test.compressedResponse {
+			// unzip the response
+			gzReader, err := gzip.NewReader(f.recorder.Body)
+			if err != nil {
+				t.Fatalf("Unexpected gzip error: %v", err)
+			}
+			plainOutput, err = io.ReadAll(gzReader)
+			if err != nil {
+				t.Fatalf("Unexpected error on reading the response: %v", err)
+			}
+		} else {
+			plainOutput = f.recorder.Body.Bytes()
+		}
+
+		expected := "true"
+		result := strings.TrimSuffix(string(plainOutput), "\n")
+		if plainOutput == nil || result != expected {
+			t.Fatalf("Expected %v but got: %v", expected, result)
+		}
+	}
+}
+
+func TestDataPostV1CompressedResponse(t *testing.T) {
+	tests := []struct {
+		gzipMinLength      int
+		compressedResponse bool
+	}{
+		{
+			gzipMinLength:      3,
+			compressedResponse: true,
+		},
+		{
+			gzipMinLength:      1400,
+			compressedResponse: false,
+		},
+	}
+
+	for _, test := range tests {
+		f := newFixtureWithConfig(t, fmt.Sprintf(`{"server":{"encoding":{"gzip":{"min_length": %d}}}}`, test.gzipMinLength))
+		// create the policy
+		err := f.v1(http.MethodPut, "/policies/test", `package test
+default hello := false
+hello {
+	input.message == "world"
+}
+`, 200, "")
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// execute the request
+		req := newReqV1(http.MethodPost, "/data/test", `{"input": {"message": "world"}}`)
+		req.Header.Set("Accept-Encoding", "gzip")
+		f.reset()
+		f.server.Handler.ServeHTTP(f.recorder, req)
+
+		var result types.DataResponseV1
+
+		// check for content encoding
+		expectedEncoding := "gzip"
+		if !test.compressedResponse {
+			expectedEncoding = ""
+		}
+		receivedEncodingHeaderValue := f.recorder.Header().Get("Content-Encoding")
+		if receivedEncodingHeaderValue != expectedEncoding {
+			t.Fatalf("Expected Content-Encoding %v but got: %v", expectedEncoding, receivedEncodingHeaderValue)
+		}
+
+		if test.compressedResponse {
+			// unzip and unmarshall the response
+			gzReader, err := gzip.NewReader(f.recorder.Body)
+			if err != nil {
+				t.Fatalf("Unexpected gzip error: %v", err)
+			}
+			if err := util.NewJSONDecoder(gzReader).Decode(&result); err != nil {
+				t.Fatalf("Unexpected JSON decode error: %v", err)
+			}
+		} else {
+			// unmarshall the response
+			if err := util.NewJSONDecoder(f.recorder.Body).Decode(&result); err != nil {
+				t.Fatalf("Unexpected JSON decode error: %v", err)
+			}
+		}
+
+		var expected interface{}
+		if err := util.UnmarshalJSON([]byte(`{"hello": true}`), &expected); err != nil {
+			panic(err)
+		}
+		if result.Result == nil || !reflect.DeepEqual(*result.Result, expected) {
+			t.Fatalf("Expected %v but got: %v", expected, *result.Result)
+		}
+	}
+}
+
+func TestCompileV1CompressedResponse(t *testing.T) {
+	tests := []struct {
+		gzipMinLength      int
+		compressedResponse bool
+	}{
+		{
+			gzipMinLength:      3,
+			compressedResponse: true,
+		},
+		{
+			gzipMinLength:      1400,
+			compressedResponse: false,
+		},
+	}
+
+	for _, test := range tests {
+		f := newFixtureWithConfig(t, fmt.Sprintf(`{"server":{"encoding":{"gzip":{"min_length": %d}}}}`, test.gzipMinLength))
+
+		// create the policy
+		mod := `package test
+
+	p {
+		input.x = 1
+	}
+
+	q {
+		data.a[i] = input.x
+	}
+
+	default r = true
+
+	r { input.x = 1 }
+
+	custom_func(x) { data.a[i] == x }
+
+	s { custom_func(input.x) }
+	`
+		err := f.v1(http.MethodPut, "/policies/test", mod, 200, "")
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// execute the request
+		req := newReqV1(http.MethodPost, "/compile", `{"unknowns": ["input"], "query": "data.test.p = true"}`)
+		req.Header.Set("Accept-Encoding", "gzip")
+		f.reset()
+		f.server.Handler.ServeHTTP(f.recorder, req)
+
+		var result types.CompileResponseV1
+
+		// check for content encoding
+		expectedEncoding := "gzip"
+		if !test.compressedResponse {
+			expectedEncoding = ""
+		}
+		receivedEncodingHeaderValue := f.recorder.Header().Get("Content-Encoding")
+		if receivedEncodingHeaderValue != expectedEncoding {
+			t.Fatalf("Expected Content-Encoding %v but got: %v", expectedEncoding, receivedEncodingHeaderValue)
+		}
+
+		if test.compressedResponse {
+			// unzip and unmarshall the response
+			gzReader, err := gzip.NewReader(f.recorder.Body)
+			if err != nil {
+				t.Fatalf("Unexpected gzip error: %v", err)
+			}
+			if err := util.NewJSONDecoder(gzReader).Decode(&result); err != nil {
+				t.Fatalf("Unexpected JSON decode error: %v", err)
+			}
+		} else {
+			// unmarshall the response
+			if err := util.NewJSONDecoder(f.recorder.Body).Decode(&result); err != nil {
+				t.Fatalf("Unexpected JSON decode error: %v", err)
+			}
+		}
+
+		var expected interface{}
+		expectedStr := fmt.Sprintf(`{"queries": [%v]}`, string(util.MustMarshalJSON(ast.MustParseBody("input.x = 1"))))
+		if err := util.UnmarshalJSON([]byte(expectedStr), &expected); err != nil {
+			panic(err)
+		}
+
+		if result.Result == nil || !reflect.DeepEqual(*result.Result, expected) {
+			t.Fatalf("Expected %v but got: %v", expected, *result.Result)
+		}
+	}
+}
+
+func TestDataPostV0CompressedRequest(t *testing.T) {
+	f := newFixture(t)
+	// create the policy
+	err := f.v1(http.MethodPut, "/policies/test", `package opa.examples
+import input.example.flag
+allow_request { flag == true }
+`, 200, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// execute the request
+	compressedBoy := zipString(`{"example": {"flag": true}}`)
+	req := newStreamedReqV0(http.MethodPost, "/data/opa/examples/allow_request", bytes.NewReader(compressedBoy))
+	req.Header.Set("Content-Encoding", "gzip")
+	f.reset()
+	f.server.Handler.ServeHTTP(f.recorder, req)
+
+	expected := "true"
+	result := strings.TrimSuffix(f.recorder.Body.String(), "\n")
+	if result != expected {
+		t.Fatalf("Expected %v but got: %v", expected, result)
+	}
+}
+
+func TestDataPostV1CompressedRequest(t *testing.T) {
+	f := newFixture(t)
+	// create the policy
+	err := f.v1(http.MethodPut, "/policies/test", `package test
+default hello := false
+hello {
+	input.message == "world"
+}
+`, 200, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// execute the request
+	compressedBoy := zipString(`{"input": {"message": "world"}}`)
+	req := newStreamedReqV1(http.MethodPost, "/data/test", bytes.NewReader(compressedBoy))
+	req.Header.Set("Content-Encoding", "gzip")
+	f.reset()
+	f.server.Handler.ServeHTTP(f.recorder, req)
+
+	var result types.DataResponseV1
+
+	// unmarshall the response
+	if err := util.NewJSONDecoder(f.recorder.Body).Decode(&result); err != nil {
+		t.Fatalf("Unexpected JSON decode error: %v", err)
+	}
+
+	var expected interface{}
+	if err := util.UnmarshalJSON([]byte(`{"hello": true}`), &expected); err != nil {
+		panic(err)
+	}
+	if result.Result == nil || !reflect.DeepEqual(*result.Result, expected) {
+		t.Fatalf("Expected %v but got: %v", expected, *result.Result)
+	}
+}
+
+func TestCompileV1CompressedRequest(t *testing.T) {
+	f := newFixture(t)
+	// create the policy
+	mod := `package test
+
+	p {
+		input.x = 1
+	}
+
+	q {
+		data.a[i] = input.x
+	}
+
+	default r = true
+
+	r { input.x = 1 }
+
+	custom_func(x) { data.a[i] == x }
+
+	s { custom_func(input.x) }
+	`
+	err := f.v1(http.MethodPut, "/policies/test", mod, 200, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// execute the request
+	compressedBoy := zipString(`{"unknowns": ["input"], "query": "data.test.p = true"}`)
+	req := newStreamedReqV1(http.MethodPost, "/compile", bytes.NewReader(compressedBoy))
+	req.Header.Set("Content-Encoding", "gzip")
+	f.reset()
+	f.server.Handler.ServeHTTP(f.recorder, req)
+
+	var result types.CompileResponseV1
+
+	// unmarshall the response
+	if err := util.NewJSONDecoder(f.recorder.Body).Decode(&result); err != nil {
+		t.Fatalf("Unexpected JSON decode error: %v", err)
+	}
+
+	var expected interface{}
+	expectedStr := fmt.Sprintf(`{"queries": [%v]}`, string(util.MustMarshalJSON(ast.MustParseBody("input.x = 1"))))
+	if err := util.UnmarshalJSON([]byte(expectedStr), &expected); err != nil {
+		panic(err)
+	}
+
+	if result.Result == nil || !reflect.DeepEqual(*result.Result, expected) {
+		t.Fatalf("Expected %v but got: %v", expected, *result.Result)
+	}
+}
+
 func TestBundleScope(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -3926,6 +4266,36 @@ func newFixture(t *testing.T, opts ...func(*Server)) *fixture {
 	}
 }
 
+func newFixtureWithConfig(t *testing.T, config string, opts ...func(*Server)) *fixture {
+	ctx := context.Background()
+	server := New().
+		WithAddresses([]string{"localhost:8182"}).
+		WithStore(inmem.New()) // potentially overridden via opts
+	for _, opt := range opts {
+		opt(server)
+	}
+
+	m, err := plugins.New([]byte(config), "test", server.store)
+	if err != nil {
+		t.Fatal(err)
+	}
+	server = server.WithManager(m)
+	if err := m.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	server, err = server.Init(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	recorder := httptest.NewRecorder()
+
+	return &fixture{
+		server:   server,
+		recorder: recorder,
+		t:        t,
+	}
+}
+
 func newFixtureWithStore(t *testing.T, store storage.Store, opts ...func(*Server)) *fixture {
 	ctx := context.Background()
 	m, err := plugins.New([]byte{}, "test", store)
@@ -4097,6 +4467,26 @@ func newReqUnversioned(method, path, body string) *http.Request {
 	return req
 }
 
+func newStreamedReqV0(method string, path string, body io.Reader) *http.Request {
+	return newStreamedReq(0, method, path, body)
+}
+
+func newStreamedReqV1(method string, path string, body io.Reader) *http.Request {
+	return newStreamedReq(1, method, path, body)
+}
+
+func newStreamedReq(version int, method string, path string, body io.Reader) *http.Request {
+	return newStreamedReqUnversioned(method, fmt.Sprintf("/v%d", version)+path, body)
+}
+
+func newStreamedReqUnversioned(method string, path string, body io.Reader) *http.Request {
+	req, err := http.NewRequest(method, path, body)
+	if err != nil {
+		panic(err)
+	}
+	return req
+}
+
 func mustUnmarshalTrace(t types.TraceV1) (trace types.TraceV1Raw) {
 	if err := json.Unmarshal(t, &trace); err != nil {
 		panic("not reached")
@@ -4478,3 +4868,15 @@ func (m mockHTTPListener) Shutdown(context.Context) error {
 func (m mockHTTPListener) Type() httpListenerType {
 	return m.t
 }
+
+func zipString(input string) []byte {
+	var b bytes.Buffer
+	gz := gzip.NewWriter(&b)
+	if _, err := gz.Write([]byte(input)); err != nil {
+		log.Fatal(err)
+	}
+	if err := gz.Close(); err != nil {
+		log.Fatal(err)
+	}
+	return b.Bytes()
+}