grpcgcp

package module
v1.6.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 5, 2025 License: Apache-2.0 Imports: 23 Imported by: 6

README

How to test Spanner integration

  1. Set GCP project id with GCP_PROJECT_ID environment variable.

     export GCP_PROJECT_ID=test-project

  2. Set service key credentials file using GOOGLE_APPLICATION_CREDENTIALS env variable.

     export GOOGLE_APPLICATION_CREDENTIALS=/service/account/credentials.json

  3. Run the tests.

     go test -v

To skip Spanner setup run

    SKIP_SPANNER=true go test -v

Documentation

Overview

Package grpcgcp provides grpc supports for Google Cloud APIs. For now it provides connection management with affinity support.

Note: "channel" is analagous to "connection" in our context.

Usage:

1. First, initialize the api configuration. There are two ways: 

1a. Create a json file defining the configuration and read it.

	// Create some_api_config.json
	{
		"channelPool": {
			"maxSize": 4,
			"maxConcurrentStreamsLowWatermark": 50
		},
		"method": [
			{
				"name": [ "/some.api.v1/Method1" ],
				"affinity": {
					"command": "BIND",
					"affinityKey": "key1"
				}
			},
			{
				"name": [ "/some.api.v1/Method2" ],
				"affinity": {
					"command": "BOUND",
					"affinityKey": "key2"
				}
			},
			{
				"name": [ "/some.api.v1/Method3" ],
				"affinity": {
					"command": "UNBIND",
					"affinityKey": "key3"
				}
			}
		]
	}

	jsonFile, err := ioutil.ReadFile("some_api_config.json")
	if err != nil {
		t.Fatalf("Failed to read config file: %v", err)
	}
	jsonCfg := string(jsonFile)

1b. Create apiConfig directly and convert it to json.

	// import (
	// 	configpb "github.com/GoogleCloudPlatform/grpc-gcp-go/grpcgcp/grpc_gcp"
	// )

	apiConfig := &configpb.ApiConfig{
		ChannelPool: &configpb.ChannelPoolConfig{
			MaxSize:                          4,
			MaxConcurrentStreamsLowWatermark: 50,
		},
		Method: []*configpb.MethodConfig{
			&configpb.MethodConfig{
				Name: []string{"/some.api.v1/Method1"},
				Affinity: &configpb.AffinityConfig{
					Command:     configpb.AffinityConfig_BIND,
					AffinityKey: "key1",
				},
			},
			&configpb.MethodConfig{
				Name: []string{"/some.api.v1/Method2"},
				Affinity: &configpb.AffinityConfig{
					Command:     configpb.AffinityConfig_BOUND,
					AffinityKey: "key2",
				},
			},
			&configpb.MethodConfig{
				Name: []string{"/some.api.v1/Method3"},
				Affinity: &configpb.AffinityConfig{
					Command:     configpb.AffinityConfig_UNBIND,
					AffinityKey: "key3",
				},
			},
		},
	}

	c, err := protojson.Marshal(apiConfig)
	if err != nil {
		t.Fatalf("cannot json encode config: %v", err)
	}
	jsonCfg := string(c)

2. Make ClientConn with specific DialOptions to enable grpc_gcp load balancer with provided configuration. And specify gRPC-GCP interceptors. 

conn, err := grpc.Dial(
	target,
	// Register and specify the grpc-gcp load balancer.
	grpc.WithDisableServiceConfig(),
	grpc.WithDefaultServiceConfig(
		fmt.Sprintf(
			`{"loadBalancingConfig": [{"%s":%s}]}`,
			grpcgcp.Name,
			jsonCfg,
		),
	),
	// Set grpcgcp interceptors.
	grpc.WithUnaryInterceptor(grpcgcp.GCPUnaryClientInterceptor),
	grpc.WithStreamInterceptor(grpcgcp.GCPStreamClientInterceptor),
)

Index

Constants

View Source 
const (
	FINE   = 90
	FINEST = 99
)
View Source 
const (
	// Name is the name of grpc_gcp balancer.
	Name = "grpc_gcp"
)
View Source 
const Version = "1.6.0-dev"

Version is the current grpcgcp version.

Variables

This section is empty.

Functions

func FromMEContext added in v1.4.0 

func FromMEContext(ctx context.Context) (string, bool)

FromMEContext returns the MultiEndpoint name stored in ctx, if any.

func GCPStreamClientInterceptor 

func GCPStreamClientInterceptor(
	ctx context.Context,
	desc *grpc.StreamDesc,
	cc *grpc.ClientConn,
	method string,
	streamer grpc.Streamer,
	opts ...grpc.CallOption,
) (grpc.ClientStream, error)

GCPStreamClientInterceptor intercepts the execution of a client streaming RPC and injects necessary information to be used by the picker.

func GCPUnaryClientInterceptor 

func GCPUnaryClientInterceptor(
	ctx context.Context,
	method string,
	req interface{},
	reply interface{},
	cc *grpc.ClientConn,
	invoker grpc.UnaryInvoker,
	opts ...grpc.CallOption,
) error

GCPUnaryClientInterceptor intercepts the execution of a unary RPC and injects necessary information to be used by the picker.

func NewGCPLogger added in v1.5.0 

func NewGCPLogger(logger grpclog.LoggerV2, prefix string) *gcpLogger

func NewMEContext added in v1.4.0 

func NewMEContext(ctx context.Context, name string) context.Context

NewMEContext returns a new Context that carries Multiendpoint name.

Types

type GCPBalancerConfig added in v1.5.0 

type GCPBalancerConfig struct {
	serviceconfig.LoadBalancingConfig
	*pb.ApiConfig
}

type GCPFallback added in v1.6.0 

type GCPFallback struct {
	grpc.ClientConnInterface
	// contains filtered or unexported fields
}

GCPFallback is a wrapper around two gRPC client connections that provides a fallback mechanism from a primary to a fallback connection based on the error rate of the primary connection.

func NewGCPFallback added in v1.6.0 

func NewGCPFallback(ctx context.Context, primaryConn grpc.ClientConnInterface, fallbackConn grpc.ClientConnInterface, fallbackOpts *GCPFallbackOptions) (*GCPFallback, error)

NewGCPFallback creates a new GCPFallback instance. It takes a primary and a fallback connection, along with options to configure the fallback behavior. GCPFallback will not close the provided connections because Close is not a part of ClientConnInterface, thus the caller is responsible for closing them properly.

func (*GCPFallback) Close added in v1.6.0 

func (f *GCPFallback) Close()

Close stops all the background goroutines and releases resources. Another way to close GCPFallback is to cancel the provided context. But both ways do not close the underlying connections. The caller must close the primary and the fallback ClientConn on their own.

func (*GCPFallback) Invoke added in v1.6.0 

func (f *GCPFallback) Invoke(ctx context.Context, method string, args any, reply any, opts ...grpc.CallOption) error

Invoke performs a unary RPC and returns after the response is received into reply.

func (*GCPFallback) NewStream added in v1.6.0 

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

NewStream begins a streaming RPC.

type GCPFallbackOptions added in v1.6.0 

type GCPFallbackOptions struct {
	// EnableFallback controls whether the fallback mechanism is enabled.
	EnableFallback bool
	// ErrorRateThreshold is the threshold for the error rate of the primary connection.
	// 1.0 means 100% error rate.
	ErrorRateThreshold float32
	// ErroneousCodes is a list of error codes that are considered erroneous.
	ErroneousCodes []codes.Code
	// Period is the interval at which the error rate is checked.
	Period time.Duration
	// MinFailedCalls is the minimum number of failed calls since last check.
	MinFailedCalls int

	// PrimaryProbingFn is the probing function for the primary connection.
	// The fallback decision is not made based only on the error rate of the
	// probing RPCs, but all RPCs.
	PrimaryProbingFn GCPFallbackProbeFn
	// FallbackProbingFn is the probing function for the fallback connection.
	FallbackProbingFn GCPFallbackProbeFn

	// PrimaryProbingInterval is the interval at which the primary connection is probed.
	PrimaryProbingInterval time.Duration
	// FallbackProbingInterval is the interval at which the fallback connection is probed.
	FallbackProbingInterval time.Duration

	// PrimaryChannelName is the name of the primary channel.
	PrimaryChannelName string
	// FallbackChannelName is the name of the fallback channel.
	FallbackChannelName string

	// MeterProvider is the OpenTelemetry meter provider.
	MeterProvider metric.MeterProvider
}

GCPFallbackOptions holds the configuration for the GCPFallback mechanism.

func NewGCPFallbackOptions added in v1.6.0 

func NewGCPFallbackOptions() *GCPFallbackOptions

NewGCPFallbackOptions creates a new GCPFallbackOptions with default values.

type GCPFallbackProbeFn added in v1.6.0 

type GCPFallbackProbeFn func(grpc.ClientConnInterface) string

GCPFallbackProbeFn defines the function signature for probing a gRPC connection. It should return a string indicating the result of the probe. Empty string means the probe was successful.

type GCPMultiEndpoint added in v1.4.0 

type GCPMultiEndpoint struct {
	grpc.ClientConnInterface
	// contains filtered or unexported fields
}

GCPMultiEndpoint holds the state of MultiEndpoints-enabled gRPC client connection.

The purposes of GCPMultiEndpoint are:

  • Fallback to an alternative endpoint (host:port) of a gRPC service when the original endpoint is completely unavailable.
  • Be able to route an RPC call to a specific group of endpoints.
  • Be able to reconfigure endpoints in runtime.

A group of endpoints is called a multiendpoint.MultiEndpoint and is essentially a list of endpoints where priority is defined by the position in the list with the first endpoint having top priority. A MultiEndpoint tracks endpoints' availability. When a MultiEndpoint is picked for an RPC call, it picks the top priority endpoint that is currently available. More information on the multiendpoint.MultiEndpoint.

GCPMultiEndpoint can have one or more MultiEndpoint identified by its name -- arbitrary string provided in the GCPMultiEndpointOptions when configuring MultiEndpoints. This name can be used to route an RPC call to this MultiEndpoint by using the NewMEContext.

GCPMultiEndpoint uses GCPMultiEndpointOptions for initial configuration. An updated configuration can be provided at any time later using [UpdateMultiEndpoints].

Example:

Let's assume we have a service with read and write operations and the following backends:

  • service.example.com -- the main set of backends supporting all operations
  • service-fallback.example.com -- read-write replica supporting all operations
  • ro-service.example.com -- read-only replica supporting only read operations

Example configuration:

  • MultiEndpoint named "default" with endpoints:

    1. service.example.com:443

    2. service-fallback.example.com:443

  • MultiEndpoint named "read" with endpoints:

    1. ro-service.example.com:443

    2. service-fallback.example.com:443

    3. service.example.com:443

With the configuration above GCPMultiEndpoint will use the "default" MultiEndpoint by default. It means that RPC calls by default will use the main endpoint and if it is not available then the read-write replica.

To offload some read calls to the read-only replica we can specify "read" MultiEndpoint in the context. Then these calls will use the read-only replica endpoint and if it is not available then the read-write replica and if it is also not available then the main endpoint.

GCPMultiEndpoint creates a grpcgcp connection pool for every unique endpoint. For the example above three connection pools will be created.

GCPMultiEndpoint implements grpc.ClientConnInterface and can be used as a grpc.ClientConn when creating gRPC clients.

func NewGCPMultiEndpoint added in v1.5.0 

func NewGCPMultiEndpoint(meOpts *GCPMultiEndpointOptions, opts ...grpc.DialOption) (*GCPMultiEndpoint, error)

NewGCPMultiEndpoint creates new GCPMultiEndpoint -- MultiEndpoints-enabled gRPC client connection.

GCPMultiEndpoint implements grpc.ClientConnInterface and can be used as a grpc.ClientConn when creating gRPC clients.

func NewGcpMultiEndpoint deprecated added in v1.4.0 

func NewGcpMultiEndpoint(meOpts *GCPMultiEndpointOptions, opts ...grpc.DialOption) (*GCPMultiEndpoint, error)

NewGCPMultiEndpoint creates new GCPMultiEndpoint -- MultiEndpoints-enabled gRPC client connection.

Deprecated: use NewGCPMultiEndpoint.

func (*GCPMultiEndpoint) Close added in v1.4.0 

func (gme *GCPMultiEndpoint) Close() error

func (*GCPMultiEndpoint) GCPConfig added in v1.5.0 

func (gme *GCPMultiEndpoint) GCPConfig() *pb.ApiConfig

func (*GCPMultiEndpoint) Invoke added in v1.4.0 

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

func (*GCPMultiEndpoint) NewStream added in v1.4.0 

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

func (*GCPMultiEndpoint) UpdateMultiEndpoints added in v1.4.0 

func (gme *GCPMultiEndpoint) UpdateMultiEndpoints(meOpts *GCPMultiEndpointOptions) error

UpdateMultiEndpoints reconfigures MultiEndpoints.

MultiEndpoints are matched with the current ones by name.

  • If a current MultiEndpoint is missing in the updated list, the MultiEndpoint will be removed.
  • A new MultiEndpoint will be created for every new name in the list.
  • For an existing MultiEndpoint only its endpoints will be updated (no recovery timeout change).

Endpoints are matched by the endpoint address (usually in the form of address:port).

  • If an existing endpoint is not used by any MultiEndpoint in the updated list, then the connection poll for this endpoint will be shutdown.
  • A connection pool will be created for every new endpoint.
  • For an existing endpoint nothing will change (the connection pool will not be re-created, thus no connection credentials change, nor connection configuration change).

type GCPMultiEndpointOptions added in v1.4.0 

type GCPMultiEndpointOptions struct {
	// Regular gRPC-GCP configuration to be applied to every endpoint.
	GRPCgcpConfig *pb.ApiConfig
	// Map of MultiEndpoints where key is the MultiEndpoint name.
	MultiEndpoints map[string]*multiendpoint.MultiEndpointOptions
	// Name of the default MultiEndpoint.
	Default string
	// Func to dial grpc ClientConn.
	DialFunc func(ctx context.Context, target string, dopts ...grpc.DialOption) (*grpc.ClientConn, error)
}

GCPMultiEndpointOptions holds options to construct a MultiEndpoints-enabled gRPC client connection.

Source Files

View all Source files

Directories

Path Synopsis
Package mocks is a generated GoMock package.
 Package mocks is a generated GoMock package.
Package multiendpoint implements multiendpoint feature.
 Package multiendpoint implements multiendpoint feature.
test_grpc
helloworld/helloworld

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.