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"
+        "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 response does not specify it,
+        // the gRPC code is derived from HTTP status code.
+        HeaderGRPCCode = "X-Prpc-Grpc-Code"
+)
+
+// 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 {
+                                return c.responseErr(ctx, res)
+                        }
+
+                        // Read the response body.
+                        buf.Reset()
+                        var body io.Reader = res.Body
+                        if res.ContentLength > 0 {
+                                buf.Grow(int(res.ContentLength))
+                                body = io.LimitReader(body, res.ContentLength)
+                        }
+                        if _, err = buf.ReadFrom(body); err != nil {
+                                return 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)
+                },
+        )
+
+        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)
+}
+
+// responseErr converts an HTTP response to a gRPC error.
+// The returned error may be wrapped with errors.WrapTransient.
+func (c *Client) responseErr(ctx context.Context, res *http.Response) error {
+        // Read first N bytes of the body. Leave empty if cannot read.
+        bodySize := c.ErrBodySize
+        if bodySize <= 0 {
+                bodySize = 256
+        }
+        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(ctx, "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
+}
+
+// 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
+}
+
+// isTransientStatus returns true if an HTTP status code indicates a transient error.
+func isTransientStatus(status int) bool {
+        return status >= 500 || status == http.StatusRequestTimeout
+}