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

common/prpc/client.go

+// 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"
+        "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"
+        "github.com/luci/luci-go/common/transport"
+)
+
+// Client can make pRPC calls.
+type Client struct {
+        Host    string   // host and optionally a port number of the target server.
+        Options *Options // if nil, DefaultOptions() are used.
+}
+
+// NewClient creates a new pRPC client with default options.
+func NewClient(host string) *Client {

[dnj, 2016/01/21 07:44:27]

This seems unnecessary, since "Client" has no private state to initialize. Let's
be cool like "http.Client"!

[nodir, 2016/01/22 00:47:24]

Done.

+        return &Client{Host: host, Options: DefaultOptions()}
+}
+
+// renderOptions copies client options and applies opts.
+func (c *Client) renderOptions(opts []grpc.CallOption) *Options {
+        var options Options
+        if c.Options != nil {
+                options = *c.Options
+        } else {
+                options = *DefaultOptions()

[dnj, 2016/01/21 07:44:28]

DefaultOptions generates a new instance anyway. Cloning it again seems
unnecessary.

var options *Options
if c.Options != nil {
  oc := *c.Options
  options = &oc
} else {
  options = DefaultOptions()
}

[nodir, 2016/01/22 00:47:24]

Done.

+        }
+        options.apply(opts)
+        return &options
+}
+
+// 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)
+        fields := logging.Fields{

[dnj, 2016/01/21 07:44:28]

Fields layer naturally via "SetFields". ctx = logging.SetFields(ctx,
logging.Fields{...}), then you basically get this for free with a lot less code.
All future calls logging.Errorf(ctx, "ohaithere") will inherit the installed
fields.

Granted it costs a Context derivation but ... who cares?

[nodir, 2016/01/22 00:47:23]

ah, I see. Done, but I don't see much  code-size savings. Am I using fields
right?

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

Looks like you are.

It's less about code size changes and more reliving the burden of management
from you. You can now assume these fields will always be there and move on with
your life.

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

ok so there is no "a lot less code"

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

Well, depends, each logging message where you don't have to copy/append and
remember fields is less code. In LogDog, I set fields up front, so that would be
a lot less. Maybe I should have said, "a lot less boilerplate?"

+                "Host":    c.Host,
+                "Service": serviceName,
+                "Method":  methodName,
+        }
+
+        // Send the request in a retry loop.
+        onRetry := func(err error, sleepTime time.Duration) {

[dnj, 2016/01/21 07:44:27]

I personally like to inline this w/ the retry.Retry call, but up to you.

[nodir, 2016/01/22 00:47:23]

Done.

+                fields := fields.Copy(logging.Fields{

[dnj, 2016/01/21 07:44:27]

(Layering fields is a lot cleaner than explicitly copying them IMO.)

[nodir, 2016/01/22 00:47:23]

Done.

+                        logging.ErrorKey: err,
+                        "SleepTime":      sleepTime,
+                })
+                fields.Warningf(ctx, "RPC failed transiently. Will retry in %s", sleepTime)
+        }
+        var client http.Client
+        client.Transport = transport.Get(ctx)

[dnj, 2016/01/21 07:44:27]

I seriously dislike embedding Transport in context. Why not just make it a field
of Client? Could even make the http.Client a field of Client.

[nodir, 2016/01/22 00:47:24]

If it is that serious, please start a discussion in the ML to remove the
transport package.
Done. Didn't embed as namelessly because didn't want to "inherit" all of its
methods.

+        err = retry.Retry(ctx, retry.TransientOnly(options.Retry()), func() error {
+                fields.Infof(ctx, "RPC %s/%s.%s", c.Host, serviceName, methodName)

[dnj, 2016/01/21 07:44:27]

This might be a bit wordy even for info level. Maybe debug?

[nodir, 2016/01/22 00:47:24]

Done. It is wordy indeed, we use info in
https://github.com/luci/luci-py/blob/master/appengine/components/components/n...

+
+                // Send the request.
+                req.Body = ioutil.NopCloser(bytes.NewBuffer(reqBody))

[dnj, 2016/01/21 07:44:27]

bytes.NewReader is more lightweight and appropriate here.

[nodir, 2016/01/22 00:47:24]

Done.

+                res, err := client.Do(req)
+                if err != nil {
+                        return errors.WrapTransient(fmt.Errorf("failed to send request: %s", err))

[dnj, 2016/01/21 07:44:28]

IMO you should handle transient errors more surgically. Maybe >= 500?

[nodir, 2016/01/22 00:47:24]

status is not available if err != nil

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

This sounds like a hard failure to me then, not a transient one.

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

Client.Do may return an error on URLFetch proxy failure
(https://godoc.org/google.golang.org/appengine/urlfetch#Transport.RoundTrip).
Without retries RPCs might be unreliable

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

Ugh okay. I really dislike that they don't give us a means to distinguish
between a flake (proxy failure) and an actually error in our request
(nonexistentdomainname.com doesn't exist) :/ Better to error on the side of
flake, I suppose.

+                }
+                defer res.Body.Close()
+
+                if options.resHeaderMetadata != nil {
+                        *options.resHeaderMetadata = metadata.MD(res.Header).Copy()
+                }
+
+                // Check response status code.
+                if res.StatusCode >= 300 {

[dnj, 2016/01/21 07:44:27]

Maybe http.StatusMultipleChoices instead of 300?

[nodir, 2016/01/22 00:47:24]

name "MultipleChoises" does not imply it is the first error status code. We
could have "firstBadStatus" constant (see also these markers in
https://golang.org/src/go/token/token.go#L22), but I believe HTTP status codes
are uniquos enough to know that bad statuses start with 300

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

Actually now that I think about it, you control the server, so you should
whitelist specific success codes. Maybe if StatusCode != OK? OK, NoContent?

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

Why couple tightly? HTTP status codes < 300 are not errors

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

If we fully control our server's response codes and only use OK, and we get
"Partial Content", "Reset Content", "IM Used", "Continue", "Switching
Protocols", or any other <300 response codes, something is wrong, aka error.

+                        return responseErr(ctx, res)
+                }
+
+                // Read and parse response message.
+                buf, err := ioutil.ReadAll(res.Body)

[dnj, 2016/01/21 07:44:27]

Since you're doing this in a retry loop, you could optimize allocation via
"bytes.Buffer{}":

// (Outside of loop)
respBuf := bytes.Buffer{}

// (Here)
respBuf.Reset()
if err := respBuf.ReadFrom(res.Body); err != nil { ... }
respBuf.Bytes()

Also if the server sends Content-Length, you can pre-allocate the Buffer:
buffer.Grow(size) ... and use an io.LimitedReader to enforce it.

[nodir, 2016/01/22 00:47:24]

Done.

+                if err != nil {
+                        return errors.WrapTransient(fmt.Errorf("failed to read response body: %s", err))

[dnj, 2016/01/21 07:44:27]

Why is a failure to read body transient? This would only happen if TCP is
interrupted or if there is an error in the content decoding, both of which seem
like non-transient.

[nodir, 2016/01/22 00:47:24]

Isn't TCP interruption transient? In other words, if we repeat the request we
may get lucky.
What decoding you are talking about? Isn't this raw bytes?

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

It depends why the interruption happened. e.g, if there was a data error in the
HTTP response, it might not be transient. Regardless, the TCP/IP protocol will
take care of any transient cases of this (data corruption, interruption, etc.).
I don't think this is an application layer transient failure case.
res.Body's reader wraps GZIP and chunk decoding from the raw HTTP response. It
will return the raw bytes on success, but can fail if either of those decodings
don't work.

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

Done

+                }
+
+                if options.resTrailerMetadata != nil {
+                        *options.resTrailerMetadata = metadata.MD(res.Trailer).Copy()
+                }
+
+                return proto.Unmarshal(buf, out) // non-transient error
+        }, onRetry)
+
+        if err != nil {
+                fields[logging.ErrorKey] = err

[dnj, 2016/01/21 07:44:27]

If you set fields, you can make this:
logging.WithError(err).Warningf(ctx, ...).

Athough this should probably be logging.Errorf. And you don't need to embed the
error in the message when you are using it as a field.

[nodir, 2016/01/22 00:47:24]

Done
Errors are quite severe: in ereporter2 land, a service owner receives a report
every hour if errors are logged. A failed RPC may be less severe than that, but
nothing prevents the user to print errors if he/she believes a failed RPC is
that severe.
I prefer to control how it actually looks in the logs, given the number of
fields we have already. I want to avoid the situation when a human has hard time
parsing actual error message from printed 3 fields.

+                fields.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)
+        if options.UserAgent != "" {
+                req.Header.Set("User-Agent", options.UserAgent)

[dnj, 2016/01/21 07:44:27]

Endpoints clients have a default user agent (e.g., "pRPC Client 1.0") and append
the optional user-supplied UserAgent string to that. WDYT?

[nodir, 2016/01/22 00:47:24]

ah right, forgot about that

+        }
+        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) error {
+        // Read first 256 bytes of the body. Leave empty if cannot read.
+        var body string
+        const maxBody = 256

[dnj, 2016/01/21 07:44:27]

Perhaps this could be a Client setting? That way, a user would have a method of
enlarging it if they actually want to see the whole response.

[nodir, 2016/01/22 00:47:24]

Done.

+        bodyBytes, err := ioutil.ReadAll(io.LimitReader(res.Body, maxBody))
+        if err == nil {
+                if len(bodyBytes) == maxBody {
+                        bodyBytes = append(bodyBytes, []byte("...")...)
+                }
+                body = string(bodyBytes)
+        }
+
+        const noCode = 0xffffffff

[dnj, 2016/01/21 07:44:27]

Rather than use a sentinel value, just use a boolean: hasCode := false

...
if !hasCode {

}

[nodir, 2016/01/22 00:47:24]

not relevant any more

+        code := codes.Code(noCode)
+        // Read explicit gRPC code.
+        const headerName = "X-Prpc-Grpc-Code"

[dnj, 2016/01/21 07:44:28]

Shared constant w/ server?

[nodir, 2016/01/22 00:47:23]

Added a constant

+        if codeHeader := res.Header.Get(headerName); codeHeader != "" {
+                if intCode, err := strconv.Atoi(codeHeader); err != nil {
+                        logging.Warningf(c, "could not parse %s header: %s", headerName, err)

[dnj, 2016/01/21 07:44:28]

Capitalize logging messages. Use fields:

logging.Fields{
  logging.ErrorKey: err,
  "headerName": headerName,
}.Warningf(c, ...)

If a code header is supplied, but the value can't be parsed, I don't think you
should default to HTTP status. I think you should use codes.Unknown.

[nodir, 2016/01/22 00:47:23]

Done
used a field for err.
I %s the header name only to use the constant. It won't change, so it should not
be in the fields.
 
I agree. Done

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

Acknowledged.

+                } else {
+                        code = codes.Code(intCode)
+                }
+        }
+        if code == noCode {
+                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
+}