blob: 75cc4b897b678d2b001f0f4802a28d904692c442 [file] [log] [blame]
// Copyright 2016 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package prpc
import (
"bytes"
"context"
"encoding/base64"
"fmt"
"io"
"net/http"
"net/url"
"os"
"strconv"
"strings"
"sync"
"time"
"golang.org/x/sync/semaphore"
"github.com/golang/protobuf/jsonpb"
"github.com/golang/protobuf/proto"
spb "google.golang.org/genproto/googleapis/rpc/status"
"google.golang.org/protobuf/types/known/anypb"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
"go.chromium.org/luci/common/clock"
"go.chromium.org/luci/common/errors"
"go.chromium.org/luci/common/logging"
"go.chromium.org/luci/common/retry"
"go.chromium.org/luci/common/retry/transient"
"go.chromium.org/luci/grpc/grpcutil"
)
const (
// HeaderGRPCCode is a name of the HTTP header that specifies the
// gRPC code in the response.
// A pRPC server must always specify it.
HeaderGRPCCode = "X-Prpc-Grpc-Code"
// HeaderStatusDetail is a name of the HTTP header that contains
// elements of google.rpc.Status.details field, one value per element,
// in the same order.
// The header value is a standard-base64 string of the encoded google.protobuf.Any,
// where the message encoding is the same as the response message encoding,
// i.e. depends on Accept request header.
HeaderStatusDetail = "X-Prpc-Status-Details-Bin"
// HeaderTimeout is HTTP header used to set pRPC request timeout.
// The single value should match regexp `\d+[HMSmun]`.
HeaderTimeout = "X-Prpc-Grpc-Timeout"
// DefaultMaxContentLength is the default maximum content length (in bytes)
// for a Client. It is 32MiB.
DefaultMaxContentLength = 32 * 1024 * 1024
)
var (
// DefaultUserAgent is default User-Agent HTTP header for pRPC requests.
DefaultUserAgent = "pRPC Client 1.4"
// ErrResponseTooBig is returned by Call when the Response's body size exceeds
// the Client's MaxContentLength limit.
ErrResponseTooBig = status.Error(codes.Unavailable, "prpc: response too big")
// ErrNoStreamingSupport is returned if a pRPC client is used to start a
// streaming RPC. They are not supported.
ErrNoStreamingSupport = status.Error(codes.Unimplemented, "prpc: no streaming support")
)
// Client can make pRPC calls.
//
// Changing fields after the first Call(...) is undefined behavior.
type Client struct {
C *http.Client // if nil, uses http.DefaultClient
Host string // host and optionally a port number of the target server
Options *Options // if nil, DefaultOptions() are used
// ErrBodySize is the number of bytes to truncate error messages from HTTP
// responses to.
//
// If non-positive, defaults to 256.
ErrBodySize int
// MaxContentLength, if > 0, is the maximum content length, in bytes, that a
// pRPC is willing to read from the server. If a larger content length is
// present in the response, ErrResponseTooBig will be returned.
//
// If <= 0, DefaultMaxContentLength will be used.
MaxContentLength int
// MaxConcurrentRequests, if > 0, limits how many requests to the server can
// execute at the same time. If 0 (default), there's no limit.
//
// If there are more concurrent Call(...) calls than the limit, excessive ones
// will block until there are execution "slots" available or the context is
// canceled. This waiting does not count towards PerRPCTimeout.
//
// The primary purpose of this mechanism is to reduce strain on the local
// network resources such as number of HTTP connections and HTTP2 streams.
// Note that it will not help with OOM problems, since blocked calls (and
// their bodies) all queue up in memory anyway.
MaxConcurrentRequests int
// EnableRequestCompression allows the client to compress requests if they
// are larger than a certain threshold.
//
// This is false by default. Use this option only with servers that understand
// compressed requests! These are Go servers built after Aug 15 2022. Python
// servers and olders Go servers would fail to parse the request with
// INVALID_ARGUMENT error.
//
// The response compression is configured independently on the server. The
// client always accepts compressed responses.
EnableRequestCompression bool
// PathPrefix is the prefix of the URL path, "<PathPrefix>/<service>/<method>"
// when making HTTP requests. If not set, defaults to "/prpc".
PathPrefix string
// Semaphore to limit concurrency, initialized in concurrencySem().
semOnce sync.Once
sem *semaphore.Weighted
// testPostHTTP is a test-installed callback that is invoked after an HTTP
// request finishes.
testPostHTTP func(context.Context, error) error
}
var _ grpc.ClientConnInterface = (*Client)(nil)
// Invoke performs a unary RPC and returns after the response is received
// into reply.
//
// It is a part of grpc.ClientConnInterface.
func (c *Client) Invoke(ctx context.Context, method string, args interface{}, reply interface{}, opts ...grpc.CallOption) error {
// 'method' looks like "/service.Name/MethodName".
parts := strings.Split(method, "/")
if len(parts) != 3 || parts[0] != "" {
return status.Errorf(codes.Internal, "prpc: not a valid method name %q", method)
}
serviceName, methodName := parts[1], parts[2]
// Inputs and outputs must be proto messages.
in, ok := args.(proto.Message)
if !ok {
return status.Errorf(codes.Internal, "prpc: bad argument type %T, not a proto", args)
}
out, ok := reply.(proto.Message)
if !ok {
return status.Errorf(codes.Internal, "prpc: bad reply type %T, not a proto", reply)
}
return c.Call(ctx, serviceName, methodName, in, out, opts...)
}
// NewStream begins a streaming RPC.
//
// It is a part of grpc.ClientConnInterface.
func (c *Client) NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, opts ...grpc.CallOption) (grpc.ClientStream, error) {
return nil, ErrNoStreamingSupport
}
// prepareOptions copies client options and applies opts.
func (c *Client) prepareOptions(opts []grpc.CallOption, serviceName, methodName string) *Options {
var options *Options
if c.Options != nil {
cpy := *c.Options
options = &cpy
} else {
options = DefaultOptions()
}
options.apply(opts)
options.host = c.Host
options.serviceName = serviceName
options.methodName = methodName
if options.UserAgent == "" {
options.UserAgent = DefaultUserAgent
}
return options
}
// Call performs a remote procedure call.
//
// Used by the generated code. Calling from multiple goroutines concurrently
// is safe.
//
// `opts` must be created by this package. Options from google.golang.org/grpc
// package are not supported. Panics if they are used.
//
// Propagates outgoing gRPC metadata provided via metadata.NewOutgoingContext.
// It will be available via metadata.FromIncomingContext on the other side.
// Similarly, if there is a deadline in the Context, it is be propagated
// to the server and applied to the context of the request handler there.
//
// Retries on internal transient errors and on gRPC codes considered transient
// by grpcutil.IsTransientCode. Logs unexpected errors (see ExpectedCode call
// option).
//
// Returns gRPC errors, perhaps with extra structured details if the server
// provided them. Context errors are converted into gRPC errors as well.
// See google.golang.org/grpc/status package.
func (c *Client) Call(ctx context.Context, serviceName, methodName string, in, out proto.Message, opts ...grpc.CallOption) error {
options := c.prepareOptions(opts, serviceName, methodName)
// Due to https://github.com/golang/protobuf/issues/745 bug
// in jsonpb handling of FieldMask, which are typically present in the
// request, not the response, do request via binary format.
options.inFormat = FormatBinary
reqBody, err := proto.Marshal(in)
if err != nil {
return status.Errorf(codes.Internal, "prpc: failed to marshal the request: %s", err)
}
switch options.AcceptContentSubtype {
case "", mtPRPCEncodingBinary:
options.outFormat = FormatBinary
case mtPRPCEncodingJSONPB:
options.outFormat = FormatJSONPB
case mtPRPCEncodingText:
return status.Errorf(codes.Internal, "prpc: text encoding for pRPC calls is not implemented")
default:
return status.Errorf(codes.Internal, "prpc: unrecognized contentSubtype %q of CallAcceptContentSubtype", options.AcceptContentSubtype)
}
resp, err := c.call(ctx, options, reqBody)
if err != nil {
return err
}
switch options.outFormat {
case FormatBinary:
err = proto.Unmarshal(resp, out)
case FormatJSONPB:
// We AllowUnknownFields because otherwise all prpc clients become tightly
// coupled to the server implementation and will break with codes.Internal
// when the server adds a new field to the response.
//
// We presume here that decoding a partial response will be easier to
// recover from in a deployment scenario than breaking all callers.
err = (&jsonpb.Unmarshaler{AllowUnknownFields: true}).Unmarshal(bytes.NewReader(resp), out)
default:
err = errors.Reason("unsupported outFormat: %s", options.outFormat).Err()
}
if err != nil {
return status.Errorf(codes.Internal, "prpc: failed to unmarshal the response: %s", err)
}
return nil
}
// CallWithFormats is like Call, but sends and returns raw data without
// marshaling it.
//
// Trims JSONPBPrefix from the response if necessary.
func (c *Client) CallWithFormats(ctx context.Context, serviceName, methodName string, in []byte, inf, outf Format, opts ...grpc.CallOption) ([]byte, error) {
options := c.prepareOptions(opts, serviceName, methodName)
if options.AcceptContentSubtype != "" {
return nil, status.Errorf(codes.Internal,
"prpc: CallAcceptContentSubtype option is not allowed with CallWithFormats "+
"because input/output formats are already specified")
}
options.inFormat = inf
options.outFormat = outf
return c.call(ctx, options, in)
}
// call implements Call and CallWithFormats.
func (c *Client) call(ctx context.Context, options *Options, in []byte) ([]byte, error) {
md, _ := metadata.FromOutgoingContext(ctx)
req, err := c.prepareRequest(options, md, in)
if err != nil {
return nil, err
}
ctx = logging.SetFields(ctx, logging.Fields{
"host": options.host,
"service": options.serviceName,
"method": options.methodName,
})
// These are populated below based on the response.
buf := &bytes.Buffer{}
contentType := ""
// Send the request in a retry loop. Use transient.Tag to propagate the retry
// signal from the loop body.
err = retry.Retry(ctx, transient.Only(options.Retry), func() (err error) {
// Note: `buf` is reset inside, it is safe to reuse it across attempts.
contentType, err = c.attemptCall(ctx, options, req, buf)
// Retry on regular transient errors and on per-RPC deadline. If this is
// a global deadline (i.e. `ctx` expired), the retry loop will just exit.
return grpcutil.WrapIfTransientOr(err, codes.DeadlineExceeded)
}, func(err error, sleepTime time.Duration) {
logging.Fields{
"sleepTime": sleepTime,
}.Warningf(ctx, "RPC failed transiently (retry in %s): %s", sleepTime, err)
})
// Parse the response content type, verify it is what we expect.
if err == nil {
switch f, formatErr := FormatFromContentType(contentType); {
case formatErr != nil:
err = status.Errorf(codes.Internal, "prpc: bad response content type %q: %s", contentType, formatErr)
case f != options.outFormat:
err = status.Errorf(codes.Internal, "prpc: output format (%q) doesn't match expected format (%q)",
f.MediaType(), options.outFormat.MediaType())
}
}
if err != nil {
// The context error is more interesting if it is present.
switch cerr := ctx.Err(); {
case cerr == context.DeadlineExceeded:
err = status.Error(codes.DeadlineExceeded, "prpc: overall deadline exceeded")
case cerr == context.Canceled:
err = status.Error(codes.Canceled, "prpc: call canceled")
case cerr != nil:
err = status.Error(codes.Unknown, cerr.Error())
}
// Unwrap the error since we wrap it in retry.Retry exclusively to attach
// a retry signal. call(...) **must** return standard unwrapped gRPC errors.
err = errors.Unwrap(err)
// Convert the error into status.Error (with Unknown code) if it wasn't
// a status before.
if status, ok := status.FromError(err); !ok {
err = status.Err()
}
// Log only on unexpected codes.
if code := status.Code(err); code != codes.Canceled {
ignore := false
for _, expected := range options.expectedCodes {
if code == expected {
ignore = true
break
}
}
if !ignore {
logging.Warningf(ctx, "RPC failed permanently: %s", err)
if options.Debug {
if code == codes.InvalidArgument && strings.Contains(err.Error(), "could not decode body") {
logging.Warningf(ctx, "Original request size: %d", len(in))
logging.Warningf(ctx, "Content-type: %s", options.inFormat.MediaType())
b64 := base64.StdEncoding.EncodeToString(in)
logging.Warningf(ctx, "Original request in base64 encoding: %s", b64)
}
}
}
}
// Do not return metadata from failed attempts.
options.resetResponseMetadata()
return nil, err
}
out := buf.Bytes()
if options.outFormat == FormatJSONPB {
out = bytes.TrimPrefix(out, bytesJSONPBPrefix)
}
return out, nil
}
// concurrencySem returns a semaphore to use to limit concurrency or nil if
// the concurrency is unlimited.
func (c *Client) concurrencySem() *semaphore.Weighted {
c.semOnce.Do(func() {
if c.MaxConcurrentRequests > 0 {
c.sem = semaphore.NewWeighted(int64(c.MaxConcurrentRequests))
}
})
return c.sem
}
// attemptCall makes one attempt at performing an RPC.
//
// Writes the raw response to the provided buffer, returns its content type.
//
// Returns gRPC errors.
func (c *Client) attemptCall(ctx context.Context, options *Options, req *http.Request, buf *bytes.Buffer) (contentType string, err error) {
// Wait until there's an execution slot available.
if sem := c.concurrencySem(); sem != nil {
if err := sem.Acquire(ctx, 1); err != nil {
return "", status.FromContextError(err).Err()
}
defer sem.Release(1)
}
// Respect PerRPCTimeout option.
now := clock.Now(ctx)
var requestDeadline time.Time
if options.PerRPCTimeout > 0 {
requestDeadline = now.Add(options.PerRPCTimeout)
}
// Does our parent Context have a deadline?
if deadline, ok := ctx.Deadline(); ok && (requestDeadline.IsZero() || deadline.Before(requestDeadline)) {
// Outer Context has a shorter deadline than our per-RPC deadline, so
// use it.
requestDeadline = deadline
} else if !requestDeadline.IsZero() {
// We have a shorter request deadline. Create a context for this attempt.
var cancel context.CancelFunc
ctx, cancel = clock.WithDeadline(ctx, requestDeadline)
defer cancel()
}
// On errors prefer the context error if the per-RPC context expired. It is
// a more consistent representation of what is happening. The other error is
// more chaotic, depending on when exactly the context expires.
defer func() {
if err != nil {
switch cerr := ctx.Err(); {
case cerr == context.DeadlineExceeded:
err = status.Error(codes.DeadlineExceeded, "prpc: attempt deadline exceeded")
case cerr == context.Canceled:
err = status.Error(codes.Canceled, "prpc: attempt canceled")
case cerr != nil:
err = status.Error(codes.Unknown, cerr.Error())
}
}
}()
// If we have a request deadline, propagate it to the server.
if !requestDeadline.IsZero() {
delta := requestDeadline.Sub(now)
if delta <= 0 {
// The request has already expired. This will likely never happen, since
// the outer Retry loop will have expired, but there is a very slight
// possibility of a race.
return "", status.Error(codes.DeadlineExceeded, "prpc: attempt deadline exceeded")
}
logging.Debugf(ctx, "RPC %s/%s.%s [deadline %s]", options.host, options.serviceName, options.methodName, delta)
req.Header.Set(HeaderTimeout, EncodeTimeout(delta))
} else {
logging.Debugf(ctx, "RPC %s/%s.%s", options.host, options.serviceName, options.methodName)
req.Header.Del(HeaderTimeout)
}
client := c.C
if client == nil {
client = http.DefaultClient
}
// Send the request.
req.Body, _ = req.GetBody()
res, err := client.Do(req.WithContext(ctx))
if err == nil {
defer func() {
// Drain the body before closing it to enable HTTP connection reuse. This
// is all best effort cleanup, don't check errors.
io.Copy(io.Discard, res.Body)
res.Body.Close()
}()
}
if c.testPostHTTP != nil {
err = c.testPostHTTP(ctx, err)
}
if err != nil {
return "", status.Errorf(codeForErr(err), "prpc: sending request: %s", err)
}
if options.resHeaderMetadata != nil {
md, err := headersIntoMetadata(res.Header)
if err != nil {
return "", status.Errorf(codes.Internal, "prpc: decoding headers: %s", err)
}
*options.resHeaderMetadata = md
}
if err := c.readResponseBody(ctx, buf, res); err != nil {
return "", err
}
if options.resTrailerMetadata != nil {
md, err := headersIntoMetadata(res.Trailer)
if err != nil {
return "", status.Errorf(codes.Internal, "prpc: decoding trailers: %s", err)
}
*options.resTrailerMetadata = md
}
// Read the RPC status (perhaps with details). This is nil on success.
err = c.readStatus(res, buf)
return res.Header.Get("Content-Type"), err
}
// readResponseBody copies the response body into dest.
//
// Returns gRPC errors. If the response body size exceeds the limits or the
// declared size, returns ErrResponseTooBig (which is also a gRPC error).
func (c *Client) readResponseBody(ctx context.Context, dest *bytes.Buffer, r *http.Response) error {
limit := c.MaxContentLength
if limit <= 0 {
limit = DefaultMaxContentLength
}
dest.Reset()
if l := r.ContentLength; l > 0 {
if l > int64(limit) {
logging.Errorf(ctx, "ContentLength header exceeds response body limit: %d > %d.", l, limit)
return ErrResponseTooBig
}
limit = int(l)
dest.Grow(limit)
}
limitedBody := io.LimitReader(r.Body, int64(limit))
if _, err := dest.ReadFrom(limitedBody); err != nil {
return status.Errorf(codeForErr(err), "prpc: reading response: %s", err)
}
// If there is more data in the body Reader, it means that the response
// size has exceeded our limit.
var probeB [1]byte
if n, err := r.Body.Read(probeB[:]); n > 0 || err != io.EOF {
logging.Errorf(ctx, "Response body limit %d exceeded.", limit)
return ErrResponseTooBig
}
return nil
}
// codeForErr decided a gRPC status code based on an http.Client error.
//
// In particular it recognizes IO timeouts and returns them as DeadlineExceeded
// code. This is necessary since it appears http.Client can sometimes fail with
// an IO timeout error even before the parent context.Context expires (probably
// has something to do with converting the context deadline into a timeout
// duration for the IO calls). When this happens, we still need to return
// DeadlineExceeded error.
func codeForErr(err error) codes.Code {
if os.IsTimeout(err) {
return codes.DeadlineExceeded
}
return codes.Internal
}
// readStatus retrieves the detailed status from the response.
//
// Puts it into a status.New(...).Err() error.
func (c *Client) readStatus(r *http.Response, bodyBuf *bytes.Buffer) error {
codeHeader := r.Header.Get(HeaderGRPCCode)
if codeHeader == "" {
if r.StatusCode >= 500 {
// It is possible that the request did not reach the pRPC server and
// that's why we don't have the code header. It's preferable to convert it
// to a gRPC status so that the client code treats the response
// appropriately.
code := codes.Internal
if r.StatusCode == http.StatusServiceUnavailable {
code = codes.Unavailable
}
return status.New(code, c.readErrorMessage(bodyBuf)).Err()
}
// Not a valid pRPC response.
return status.Errorf(codes.Internal,
"prpc: no %s header, HTTP status %d, body: %q",
HeaderGRPCCode, r.StatusCode, c.readErrorMessage(bodyBuf))
}
code, err := strconv.Atoi(codeHeader)
if err != nil {
return status.Errorf(codes.Internal, "prpc: invalid %s header value %q", HeaderGRPCCode, codeHeader)
}
if codes.Code(code) == codes.OK {
return nil
}
sp := &spb.Status{
Code: int32(code),
Message: strings.TrimSuffix(c.readErrorMessage(bodyBuf), "\n"),
}
if sp.Details, err = c.readStatusDetails(r); err != nil {
return err
}
return status.FromProto(sp).Err()
}
// readErrorMessage reads an error message from a body buffer.
//
// Respects c.ErrBodySize. If the error message is too long, trims it and
// appends "...".
func (c *Client) readErrorMessage(bodyBuf *bytes.Buffer) string {
ret := bodyBuf.Bytes()
// Apply limits.
limit := c.ErrBodySize
if limit <= 0 {
limit = 256
}
if len(ret) > limit {
return strings.ToValidUTF8(string(ret[:limit]), "") + "..."
}
return string(ret)
}
// readStatusDetails reads google.rpc.Status.details from the response headers.
//
// Returns gRPC errors.
func (c *Client) readStatusDetails(r *http.Response) ([]*anypb.Any, error) {
values := r.Header[HeaderStatusDetail]
if len(values) == 0 {
return nil, nil
}
ret := make([]*anypb.Any, len(values))
var buf []byte
for i, v := range values {
sz := base64.StdEncoding.DecodedLen(len(v))
if cap(buf) < sz {
buf = make([]byte, sz)
}
n, err := base64.StdEncoding.Decode(buf[:sz], []byte(v))
if err != nil {
return nil, status.Errorf(codes.Internal, "prpc: invalid header %s: %q", HeaderStatusDetail, v)
}
msg := &anypb.Any{}
if err := proto.Unmarshal(buf[:n], msg); err != nil {
return nil, status.Errorf(codes.Internal, "prpc: failed to unmarshal status detail: %s", err)
}
ret[i] = msg
}
return ret, nil
}
// prepareRequest creates an HTTP request for an RPC.
//
// Initializes GetBody, so that the request can be resent multiple times when
// retrying.
func (c *Client) prepareRequest(options *Options, md metadata.MD, requestMessage []byte) (*http.Request, error) {
// Convert metadata into HTTP headers in canonical form (i.e. Title-Case).
// Extract Host header, it is special and must be passed via
// http.Request.Host. Preallocate 5 more slots (for 4 headers below and for
// the RPC deadline header).
headers := make(http.Header, len(md)+5)
if err := metaIntoHeaders(md, headers); err != nil {
return nil, status.Errorf(codes.Internal, "prpc: headers: %s", err)
}
hostHdr := headers.Get("Host")
headers.Del("Host")
// Add protocol-related headers.
headers.Set("Content-Type", options.inFormat.MediaType())
headers.Set("Accept", options.outFormat.MediaType())
headers.Set("User-Agent", options.UserAgent)
body := requestMessage
if c.EnableRequestCompression && len(requestMessage) > gzipThreshold {
headers.Set("Content-Encoding", "gzip")
var err error
if body, err = compressBlob(requestMessage); err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
// Do not add "Accept-Encoding: gzip". The http package does this
// automatically, and also decompresses the response.
}
headers.Set("Content-Length", strconv.Itoa(len(body)))
scheme := "https"
if options.Insecure {
scheme = "http"
}
pathPrefix := c.PathPrefix
if c.PathPrefix == "" {
pathPrefix = "/prpc"
}
return &http.Request{
Method: "POST",
URL: &url.URL{
Scheme: scheme,
Host: options.host,
Path: fmt.Sprintf("%s/%s/%s", pathPrefix, options.serviceName, options.methodName),
},
Host: hostHdr,
Header: headers,
ContentLength: int64(len(body)),
GetBody: func() (io.ReadCloser, error) {
return io.NopCloser(bytes.NewReader(body)), nil
},
}, nil
}