common/prpc, tools/cmd/cproto: prpc client

common/prpc/client.go

Index: common/prpc/client.go
diff --git a/common/prpc/client.go b/common/prpc/client.go
new file mode 100644
index 0000000000000000000000000000000000000000..605967a2434c5dcd912d8521f750d3b59ebe0469
--- /dev/null
+++ b/common/prpc/client.go
@@ -0,0 +1,234 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package prpc
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"strconv"
+	"time"
+
+	"github.com/golang/protobuf/proto"
+	"golang.org/x/net/context"
+	"golang.org/x/net/context/ctxhttp"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+
+	"github.com/luci/luci-go/common/errors"
+	"github.com/luci/luci-go/common/logging"
+	"github.com/luci/luci-go/common/retry"
+)
+
+const (
+	// HeaderGrpcCode is a name of the HTTP header that specifies the
+	// gRPC code in the response. If not response does not specify it,
+	// the gRPC code is derived from HTTP status code.
+	HeaderGrpcCode = "X-Prpc-Grpc-Code"

[dnj, 2016/01/22 01:16:27]

nit: HeaderGRPCCode?

[nodir, 2016/01/22 02:57:50]

Done.

+)
+
+// DefaultUserAgent is default User-Agent HTTP header for pRPC requests.
+var DefaultUserAgent = "pRPC Client 1.0"
+
+// Client can make pRPC calls.
+type Client struct {
+	C *http.Client // if nil, uses http.DefaultClient
+
+	// ErrBodySize is the number of bytes to read from a HTTP response
+	// with status >= 300 and include in the error.
+	// If non-positive, defaults to 256.
+	ErrBodySize int
+
+	Host    string   // host and optionally a port number of the target server.
+	Options *Options // if nil, DefaultOptions() are used.
+}
+
+// renderOptions copies client options and applies opts.
+func (c *Client) renderOptions(opts []grpc.CallOption) *Options {
+	var options *Options
+	if c.Options != nil {
+		cpy := *c.Options
+		options = &cpy
+	} else {
+		options = DefaultOptions()
+	}
+	options.apply(opts)
+	return options
+}
+
+func (c *Client) getHTTPClient() *http.Client {
+	if c.C == nil {
+		return http.DefaultClient
+	}
+	return c.C
+}
+
+// Call makes an RPC.
+// Retries on transient errors according to retry options.
+// Logs HTTP errors.
+//
+// opts must be created by this package.
+// Calling from multiple goroutines concurrently is safe, unless Client is mutated.
+// Called from generated code.
+func (c *Client) Call(ctx context.Context, serviceName, methodName string, in, out proto.Message, opts ...grpc.CallOption) error {
+	options := c.renderOptions(opts)
+
+	reqBody, err := proto.Marshal(in)
+	if err != nil {
+		return err
+	}
+
+	req := prepareRequest(c.Host, serviceName, methodName, len(reqBody), options)
+	ctx = logging.SetFields(ctx, logging.Fields{
+		"Host":    c.Host,
+		"Service": serviceName,
+		"Method":  methodName,
+	})
+
+	// Send the request in a retry loop.
+	var iter retry.Iterator
+	if options.Retry != nil {
+		iter = retry.TransientOnly(options.Retry())
+	}
+	var buf bytes.Buffer
+	err = retry.Retry(
+		ctx,
+		iter,
+		func() error {
+			logging.Debugf(ctx, "RPC %s/%s.%s", c.Host, serviceName, methodName)
+
+			// Send the request.
+			req.Body = ioutil.NopCloser(bytes.NewReader(reqBody))
+			res, err := ctxhttp.Do(ctx, c.getHTTPClient(), req)
+			if err != nil {
+				return errors.WrapTransient(fmt.Errorf("failed to send request: %s", err))
+			}
+			defer res.Body.Close()
+
+			if options.resHeaderMetadata != nil {
+				*options.resHeaderMetadata = metadata.MD(res.Header).Copy()
+			}
+
+			// Check response status code.
+			if res.StatusCode >= 300 {
+				bodySize := c.ErrBodySize

[dnj, 2016/01/22 01:16:27]

Move this logic into responseErr and make responseErr a Client member function?

[nodir, 2016/01/22 02:57:50]

Done.

+				if bodySize <= 0 {
+					bodySize = 256
+				}
+				return responseErr(ctx, res, bodySize)
+			}
+
+			// Read the response body.
+			buf.Reset()
+			var body io.Reader = res.Body
+			if req.ContentLength > 0 {
+				buf.Grow(int(req.ContentLength))

[dnj, 2016/01/22 01:16:27]

<res>.ContentLength** (and above)

[nodir, 2016/01/22 02:57:50]

oops done

+				body = io.LimitReader(body, res.ContentLength)
+			}
+			if _, err = io.Copy(&buf, body); err != nil {

[dnj, 2016/01/22 01:16:27]

io.Copy reads in fixed-size chunks. You should use buf.ReadFrom, which reads in
buf-sized chunks if available. This would result in potentially zero
reallocations if the ContentLength was supplied and better performance in retry
cases.

[nodir, 2016/01/22 02:57:50]

Done.

+				return errors.WrapTransient(fmt.Errorf("failed to read response body: %s", err))
+			}
+
+			// Unmarshal the response message.
+			if options.resTrailerMetadata != nil {
+				*options.resTrailerMetadata = metadata.MD(res.Trailer).Copy()
+			}
+
+			return proto.Unmarshal(buf.Bytes(), out) // non-transient error
+		},
+		func(err error, sleepTime time.Duration) {
+			logging.Fields{
+				logging.ErrorKey: err,
+				"SleepTime":      sleepTime,
+			}.Warningf(ctx, "RPC failed transiently. Will retry in %s", sleepTime)

[dnj, 2016/01/22 01:16:27]

nit: Since you're emitting sleepTime as a Field, you don't need to repeat it in
the error string.

[nodir, 2016/01/22 02:57:50]

I don't like how fields are printed and consider them auxilliary to the log
message. My position is to have a nice log message first, and secondly extra
data in the fields. In other words, the fact that something is present in the
fields should be the reason to exclude it from the log message.

[dnj (Google), 2016/01/22 04:06:23]

The purpose of "fields" is to remove useful data from the log message string so
that it's concentrated and (potentially) exportable. They shouldn't be
considered auxiliary, just slightly more structured than a raw string. But up to
you.

+		},
+	)
+
+	if err != nil {
+		logging.WithError(err).Warningf(ctx, "RPC failed permanently: %s", err)
+	}
+
+	// We have to unwrap gRPC errors because
+	// grpc.Code and grpc.ErrorDesc functions do not work with error wrappers.
+	// https://github.com/grpc/grpc-go/issues/494
+	return errors.UnwrapAll(err)
+}
+
+// prepareRequest creates an HTTP request for an RPC,
+// except it does not set the request body.
+func prepareRequest(host, serviceName, methodName string, contentLength int, options *Options) *http.Request {
+	if host == "" {
+		panic("Host is not set")
+	}
+	req := &http.Request{
+		Method: "POST",
+		URL: &url.URL{
+			Scheme: "https",
+			Host:   host,
+			Path:   fmt.Sprintf("/prpc/%s/%s", serviceName, methodName),
+		},
+		Header: http.Header{},
+	}
+	if options.Insecure {
+		req.URL.Scheme = "http"
+	}
+
+	// Set headers.
+	const mediaType = "application/prpc" // binary
+	req.Header.Set("Content-Type", mediaType)
+	req.Header.Set("Accept", mediaType)
+	userAgent := options.UserAgent
+	if userAgent == "" {
+		userAgent = DefaultUserAgent
+	}
+	req.Header.Set("User-Agent", userAgent)
+	req.ContentLength = int64(contentLength)
+	req.Header.Set("Content-Length", strconv.Itoa(contentLength))
+	// TODO(nodir): add "Accept-Encoding: gzip" when pRPC server supports it.
+	return req
+}
+
+// responseErr converts an HTTP response to a gRPC error.
+// The returned error may be wrapped with errors.WrapTransient.
+func responseErr(c context.Context, res *http.Response, bodySize int) error {
+	// Read first 256 bytes of the body. Leave empty if cannot read.
+	var body string
+	bodyBytes, err := ioutil.ReadAll(io.LimitReader(res.Body, int64(bodySize)))
+	if err == nil {
+		if len(bodyBytes) == bodySize {
+			bodyBytes = append(bodyBytes, []byte("...")...)
+		}
+		body = string(bodyBytes)
+	}
+
+	code := codes.Unknown
+	// Read explicit gRPC code.
+	if codeHeader := res.Header.Get(HeaderGrpcCode); codeHeader != "" {
+		if intCode, err := strconv.Atoi(codeHeader); err != nil {
+			logging.WithError(err).Warningf(c, "Could not parse %s header: %s", HeaderGrpcCode, err)
+		} else {
+			code = codes.Code(intCode)
+		}
+	} else {
+		// No explicit gRPC code provided, derive it from status code.
+		code = StatusCode(res.StatusCode)
+	}
+
+	// Return the error.
+	err = grpc.Errorf(code, "HTTP %s: %s", res.Status, body)
+	if isTransientStatus(res.StatusCode) {
+		err = errors.WrapTransient(err)
+	}
+	return err
+}
+
+// isTransientStatus returns true if an HTTP status code indicates a transient error.
+func isTransientStatus(status int) bool {
+	return status >= 500 || status == http.StatusRequestTimeout
+}