dgrpc

README

StreamingFast GRPC Library

This is a helper library for instanciating gRPC clients and servers. It is part of StreamingFast.

Usage

See examples usage in examples folder.

Logging

If you define the environment variable GRPC_REGISTER_ZAP_LOGGER, the dgrpc when imported will immediately replace the standard golang.google.com/grpc/grpclog package so that grpc-go logging goes through StreamingFast standard github.com/streamingfast/logging library and you can then dynamically configure the logger level using DLOG spec.

The default registered logger has short name grpc and its id is google.golang.org/grpc.

If GRPC_REGISTER_ZAP_LOGGER was set on startup (any value is accepted), you should see rapidly gRPC internal logs printed to your console.

[!NOTE] We do not register the zap logger into gRPC stack on program initialization because it prints a lot of warning which would most probably clutter the logs by default. It make sense to set GRPC_REGISTER_ZAP_LOGGER by default if you provide a spec that ignores logs from google.golang.org/grpc by default.

Once GRPC_REGISTER_ZAP_LOGGER is set, use standard logging spec like DLOG="google.golang.org/grpc=debug" and you will see logs appearing. At any moment you can call dgrpc.SetGRPCLogger (or dgrpc.SetGRPCLoggerWithVerbosity) to override the active grpc-go *zap.Logger with your own.

Notes

As of commit 49c1ad3ecbaa5ab09850bdd555cc6d8422b6911b, the http handler for connect-web/server has changed from net/http to github.com/gorilla/mux. Please refer to https://github.com/gorilla/mux when creating a handler.

Here is an example on how to do it:

options := []dgrpcserver.Option{
    [...]
}
options = append(options,
    dgrpcserver.WithConnectWebHTTPHandlers([]dgrpcserver.HTTPHandlerGetter{
        func() (string, http.Handler) {
            return "/auth/callback", authCallback
        },

        func() (string, http.Handler) {
            return "/articles/{id}/{articleName}", articlesCallback
        },
    }),
)
[...]

srv := connectweb.New([]connectweb.HandlerGetter{...httpHandlers}, options...)
[...]
srv.Launch(addr)

Contributing

Issues and PR in this repo related strictly to the dgrpc library.

Report any protocol-specific issues in their respective repositories

Please first refer to the general StreamingFast contribution guide, if you wish to contribute to this code base.

This codebase uses unit tests extensively, please write and run tests.

License

Apache 2.0

Documentation

Index

• func AsGRPCError(err error) *status.Status
• func GetXDSBootstrapFilename() string
• func IsGRPCErrorCode(err error, code codes.Code) bool
• func IsXDSRemoteAddr(remoteAddr string) bool
• func NewClientConn(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)
• func NewExternalClient(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)
• func NewExternalClientConn(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)
• func NewInternalClient(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)
• func NewInternalClientConn(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)
• func NewInternalNoWaitClient(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)
• func NewInternalNoWaitClientConn(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)
• func RegisterKubernetesResolver(customScheme string)
• func SetGRPCLogger(logger *zap.Logger, tracer logging.Tracer)
• func SetGRPCLoggerWithVerbosity(logger *zap.Logger, tracer logging.Tracer, verbosity int)
• type RoundRobinConnPool
  • func NewRoundRobinConnPool(conns []*grpc.ClientConn) *RoundRobinConnPool
  • func (p *RoundRobinConnPool) Close() error
  • func (p *RoundRobinConnPool) Conn() *grpc.ClientConn
  • func (p *RoundRobinConnPool) Invoke(ctx context.Context, method string, args interface{}, reply interface{}, ...) error
  • func (p *RoundRobinConnPool) NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, ...) (grpc.ClientStream, error)
  • func (p *RoundRobinConnPool) Num() int
• type SettableLoggerV2

Functions

func AsGRPCError

func AsGRPCError(err error) *status.Status

AsGRPCError recursively finds the first value [Status] representation out of this error stack. Refers to status.FromError for details how this is tested.

If no such [Status] can be found, nil is returned, expected usage is:

if err := AsGRPCError(err); err != nil && err.Code == codes.Canceled {
	// Do something
}

func GetXDSBootstrapFilename

func GetXDSBootstrapFilename() string

GetXDSBootstrapFilename returns the filename of the xDS bootstrap file which is currently simply returning the value of the 'GRPC_XDS_BOOTSTRAP' environment variable.

We might add more logic to this function in the future.

func IsGRPCErrorCode

func IsGRPCErrorCode(err error, code codes.Code) bool

IsGRPCErrorCode is a convenience to reduce code when using AsGRPCError:

if err := AsGRPCError(err); err != nil && err.Code() == code {
	return true
}

return false

func IsXDSRemoteAddr

func IsXDSRemoteAddr(remoteAddr string) bool

IsXDSRemoteAddr returns true if the remote address is an xDS address (i.e. starts with "xds://").

func NewClientConn

func NewClientConn(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)

NewClientConn creates a default gRPC ClientConn with round robin, keepalive, OpenTelemetry tracing and large receive message size (max 1 GiB) configured.

If the remoteAddr starts with "xds://", it will use xDS credentials, transport credentials are not configured and default gRPC applies which is full TLS.

It accepts extra gRPC DialOptions to be passed to the grpc.Dial function.

func NewExternalClient

func NewExternalClient(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)

Deprecated: use NewExternalClientConn instead

func NewExternalClientConn

func NewExternalClientConn(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)

NewExternalClientConn creates a default gRPC ClientConn with round robin, keepalive, OpenTelemetry tracing, large receive message size (max 1 GiB) and TLS secure credentials configured.

func NewInternalClient

func NewInternalClient(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)

Deprecated: use NewInternalClientConn instead

func NewInternalClientConn

func NewInternalClientConn(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)

NewInternalClientConn creates a grpc ClientConn with keep alive, tracing and plain text connection (so no TLS involved, the server must also listen to a plain text socket). InternalClient also has the default call option to "WaitForReady", which means that it will hang indefinitely if the provided remote address does not resolve to any valid endpoint. This is a desired behavior for internal services managed by "discovery service" mechanisms where the remote endpoint may become ready soon.

It's possible to debug low-level message using `export GODEBUG=http2debug=2`.

func NewInternalNoWaitClient

func NewInternalNoWaitClient(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)

Deprecated: use NewInternalNoWaitClientConn instead

func NewInternalNoWaitClientConn

func NewInternalNoWaitClientConn(remoteAddr string, extraOpts ...grpc.DialOption) (*grpc.ClientConn, error)

NewInternalNoWaitClientConn creates a grpc ClientConn with keep alive, tracing and plain text connection (so no TLS involved, the server must also listen to a plain text socket). InternalClient does not have the default call option to "WaitForReady", which means that it will not hang indefinitely if the provided remote address does not resolve to any valid endpoint. This is a desired behavior for internal services where the remote endpoint is not managed by a "discovery service" mechanism.

func RegisterKubernetesResolver

func RegisterKubernetesResolver(customScheme string)

RegisterKubernetesResolver registers the "kubernetes" resolver with the given scheme. Refers to github.com/sercand/kuberesolver/v5 for more information about the `kubernetes:///` gRPC resolver.

**Important** The 'dgrpc' library already registers the kubernetes resolver with the 'kubernetes' scheme. so it's not necessary to call this function unless you want to use a different scheme. You can define the variable `GRPC_REGISTER_KUBERNETES_RESOLVER` to `false` to prevent the automatic registration. if needed.

func SetGRPCLogger

func SetGRPCLogger(logger *zap.Logger, tracer logging.Tracer)

SetGRPCLogger replaces the grpc_log.LoggerV2 with the provided logger. The gRPC level logging verbosity which is a value between 0 and 3 (both end inclusive) is inferred based on the logger/tracer level. If logger is in INFO or higher (WARN, ERROR, DPANIC, PANIC, FATAL), the verbosity is set to 0. If the logger is in DEBUG, the verbosity is set to 1. If the logger is in TRACE (tracer.Enabled()), the verbosity is set to 3 (full verbosity).

Use SetGRPCLoggerWithVerbosity to control the verbosity manually.

If you don't have a 'tracer', you can pass nil here.

func SetGRPCLoggerWithVerbosity

func SetGRPCLoggerWithVerbosity(logger *zap.Logger, tracer logging.Tracer, verbosity int)

SetGrpcLoggerV2WithVerbosity replaces the grpc_.LoggerV2 with the provided logger and verbosity. It can be used even when grpc infrastructure was initialized.

If you don't have a 'tracer', you can pass nil here.

Types

type RoundRobinConnPool

type RoundRobinConnPool struct {
	// contains filtered or unexported fields
}

func NewRoundRobinConnPool

func NewRoundRobinConnPool(conns []*grpc.ClientConn) *RoundRobinConnPool

func (*RoundRobinConnPool) Close

func (p *RoundRobinConnPool) Close() error

func (*RoundRobinConnPool) Conn

func (p *RoundRobinConnPool) Conn() *grpc.ClientConn

func (*RoundRobinConnPool) Invoke

func (p *RoundRobinConnPool) Invoke(ctx context.Context, method string, args interface{}, reply interface{}, opts ...grpc.CallOption) error

func (*RoundRobinConnPool) NewStream

func (p *RoundRobinConnPool) NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, opts ...grpc.CallOption) (grpc.ClientStream, error)

func (*RoundRobinConnPool) Num

func (p *RoundRobinConnPool) Num() int

type SettableLoggerV2

type SettableLoggerV2 interface {
	grpclog.LoggerV2
	// Sets given logger as the underlying implementation.
	Set(loggerv2 grpclog.LoggerV2)
	// Sets `discard` logger as the underlying implementation.
	Reset()
}

SettableLoggerV2 is thread-safe.