Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Issue 2241: support compression #2843

Merged
merged 20 commits into from
Dec 27, 2018
Merged

Issue 2241: support compression #2843

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
f943830
Allow enabling compression with env var when live loading.
Dec 15, 2018
d752441
Merge changes from master.
Dec 17, 2018
8fd6fff
Enable gzipped response over HTTP.
Dec 18, 2018
1394b9a
Add test for query response compression.
Dec 19, 2018
1b8433f
Support gzipped request over HTTP.
Dec 19, 2018
fafa5dc
Add test for compressed HTTP request.
Dec 19, 2018
72ab9a1
Merge in master.
Dec 19, 2018
5795cf6
Add --use_compression option with live loader.
Dec 20, 2018
03ca90c
Always use no compression with zero server.
Dec 20, 2018
dfbe54c
Merge branch 'master' into javier/issue2241_support_compression
Dec 21, 2018
6e28a65
Minor wording change.
Dec 21, 2018
1831c2d
Minor wording change.
Dec 21, 2018
316c1cc
Add simple test for gRPC compression support.
Dec 21, 2018
6ce90e0
Fix calls to connection setup.
Dec 21, 2018
7800f43
Better test function name.
Dec 21, 2018
bcffff2
Refactor HTTP handlers to share common functionality instead of dupli…
Dec 24, 2018
7d5c3a3
Merge branch 'master' into javier/issue2241_support_compression
Dec 24, 2018
b18b689
Remove debug code & minor cleanup.
Dec 24, 2018
0b96034
Fixes from PR review.
Dec 27, 2018
5c5b906
Merge branch 'master' into javier/issue2241_support_compression
Dec 27, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
157 changes: 88 additions & 69 deletions dgraph/cmd/alpha/http.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,11 @@ package alpha

import (
"bytes"
"compress/gzip"
"context"
"encoding/json"
"fmt"
"io"
"io/ioutil"
"net/http"
"sort"
Expand All @@ -34,7 +36,9 @@ import (
"github.com/dgraph-io/dgraph/query"
"github.com/dgraph-io/dgraph/worker"
"github.com/dgraph-io/dgraph/x"

"github.com/golang/glog"

"google.golang.org/grpc/metadata"
)

Expand All @@ -60,19 +64,71 @@ func extractStartTs(urlPath string) (uint64, error) {
}
}

// This method should just build the request and proxy it to the Query method of dgraph.Server.
// It can then encode the response as appropriate before sending it back to the user.
func queryHandler(w http.ResponseWriter, r *http.Request) {
// Common functionality for these request handlers. Returns true if the request is completely
// handled here and nothing further needs to be done.
func commonHandler(w http.ResponseWriter, r *http.Request) bool {
// Do these requests really need CORS headers? Doesn't seem like it, but they are probably
// harmless aside from the extra size they add to each response.
x.AddCorsHeaders(w)
w.Header().Set("Content-Type", "application/json")

if r.Method == "OPTIONS" {
return
}

if !allowed(r.Method) {
return true
} else if !allowed(r.Method) {
w.WriteHeader(http.StatusBadRequest)
x.SetStatus(w, x.ErrorInvalidMethod, "Invalid method")
return true
}

return false
}

// Read request body, transparently decompressing if necessary. Return nil on error.
func readRequest(w http.ResponseWriter, r *http.Request) []byte {
var in io.Reader = r.Body

if enc := r.Header.Get("Content-Encoding"); enc != "" && enc != "identity" {
if enc == "gzip" {
gz, err := gzip.NewReader(r.Body)
if err != nil {
x.SetStatus(w, x.Error, "Unable to create decompressor")
return nil
}
defer gz.Close()
in = gz
} else {
x.SetStatus(w, x.ErrorInvalidRequest, "Unsupported content encoding")
return nil
}
}

body, err := ioutil.ReadAll(in)
if err != nil {
x.SetStatus(w, x.ErrorInvalidRequest, err.Error())
return nil
}

return body
}

// Write response body, transparently compressing if necessary.
func writeResponse(w http.ResponseWriter, r *http.Request, b []byte) (int, error) {
var out io.Writer = w

if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
w.Header().Set("Content-Encoding", "gzip")
gzw := gzip.NewWriter(w)
defer gzw.Close()
out = gzw
}

return out.Write(b)
}

// This method should just build the request and proxy it to the Query method of dgraph.Server.
// It can then encode the response as appropriate before sending it back to the user.
func queryHandler(w http.ResponseWriter, r *http.Request) {
if commonHandler(w, r) {
return
}

Expand All @@ -93,13 +149,11 @@ func queryHandler(w http.ResponseWriter, r *http.Request) {
}
}

defer r.Body.Close()
q, err := ioutil.ReadAll(r.Body)
if err != nil {
x.SetStatus(w, x.ErrorInvalidRequest, err.Error())
body := readRequest(w, r)
if body == nil {
return
}
req.Query = string(q)
req.Query = string(body)

d := r.URL.Query().Get("debug")
ctx := context.WithValue(context.Background(), "debug", d)
Expand Down Expand Up @@ -159,32 +213,24 @@ func queryHandler(w http.ResponseWriter, r *http.Request) {
writeEntry("data", resp.Json)
}
out.WriteRune('}')
w.Write(out.Bytes())

writeResponse(w, r, out.Bytes())
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Error return value of writeResponse is not checked (from errcheck)

}

func mutationHandler(w http.ResponseWriter, r *http.Request) {
x.AddCorsHeaders(w)
w.Header().Set("Content-Type", "application/json")

if r.Method == "OPTIONS" {
if commonHandler(w, r) {
return
}

if !allowed(r.Method) {
w.WriteHeader(http.StatusBadRequest)
x.SetStatus(w, x.ErrorInvalidMethod, "Invalid method")
return
}
defer r.Body.Close()
m, err := ioutil.ReadAll(r.Body)
if err != nil {
x.SetStatus(w, x.ErrorInvalidRequest, err.Error())
m := readRequest(w, r)
if m == nil {
return
}

parseStart := time.Now()

var mu *api.Mutation
var err error
if mType := r.Header.Get("X-Dgraph-MutationType"); mType == "json" {
// Parse JSON.
ms := make(map[string]*skipJSONUnmarshal)
Expand Down Expand Up @@ -261,20 +307,12 @@ func mutationHandler(w http.ResponseWriter, r *http.Request) {
x.SetStatusWithData(w, x.Error, err.Error())
return
}
w.Write(js)

writeResponse(w, r, js)
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Error return value of writeResponse is not checked (from errcheck)

}

func commitHandler(w http.ResponseWriter, r *http.Request) {
x.AddCorsHeaders(w)
w.Header().Set("Content-Type", "application/json")

if r.Method == "OPTIONS" {
return
}

if !allowed(r.Method) {
w.WriteHeader(http.StatusBadRequest)
x.SetStatus(w, x.ErrorInvalidMethod, "Invalid method")
if commonHandler(w, r) {
return
}

Expand All @@ -296,10 +334,8 @@ func commitHandler(w http.ResponseWriter, r *http.Request) {
tc.StartTs = ts

// Keys are sent as an array in the body.
defer r.Body.Close()
keys, err := ioutil.ReadAll(r.Body)
if err != nil {
x.SetStatus(w, x.ErrorInvalidRequest, err.Error())
keys := readRequest(w, r)
if keys == nil {
return
}

Expand Down Expand Up @@ -335,20 +371,12 @@ func commitHandler(w http.ResponseWriter, r *http.Request) {
x.SetStatusWithData(w, x.Error, err.Error())
return
}
w.Write(js)

writeResponse(w, r, js)
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Error return value of writeResponse is not checked (from errcheck)

}

func abortHandler(w http.ResponseWriter, r *http.Request) {
x.AddCorsHeaders(w)
w.Header().Set("Content-Type", "application/json")

if r.Method == "OPTIONS" {
return
}

if !allowed(r.Method) {
w.WriteHeader(http.StatusBadRequest)
x.SetStatus(w, x.ErrorInvalidMethod, "Invalid method")
if commonHandler(w, r) {
return
}

Expand Down Expand Up @@ -385,33 +413,23 @@ func abortHandler(w http.ResponseWriter, r *http.Request) {
x.SetStatusWithData(w, x.Error, err.Error())
return
}
w.Write(js)

writeResponse(w, r, js)
}

func alterHandler(w http.ResponseWriter, r *http.Request) {
x.AddCorsHeaders(w)
w.Header().Set("Content-Type", "application/json")

if r.Method == "OPTIONS" {
return
}

if !allowed(r.Method) {
w.WriteHeader(http.StatusBadRequest)
x.SetStatus(w, x.ErrorInvalidMethod, "Invalid method")
if commonHandler(w, r) {
return
}

op := &api.Operation{}

defer r.Body.Close()
b, err := ioutil.ReadAll(r.Body)
if err != nil {
x.SetStatus(w, x.ErrorInvalidRequest, err.Error())
b := readRequest(w, r)
if b == nil {
return
}

err = json.Unmarshal(b, &op)
err := json.Unmarshal(b, &op)
if err != nil {
op.Schema = string(b)
}
Expand Down Expand Up @@ -442,7 +460,8 @@ func alterHandler(w http.ResponseWriter, r *http.Request) {
x.SetStatus(w, x.Error, err.Error())
return
}
w.Write(js)

writeResponse(w, r, js)
}

// skipJSONUnmarshal stores the raw bytes as is while JSON unmarshaling.
Expand Down
Loading