transport

Documentation

Overview

Package transport implements the xDS transport protocol functionality required by the xdsclient.

Index

• type ADSFlowControl
  • func NewADSStreamFlowControl() *ADSFlowControl
  • func (fc *ADSFlowControl) Add()
  • func (fc *ADSFlowControl) OnDone()
  • func (fc *ADSFlowControl) Wait(ctx context.Context) bool
• type OnRecvHandlerFunc
• type OnSendHandlerFunc
• type Options
• type ResourceSendInfo
• type ResourceUpdate
• type Transport
  • func New(opts Options) (*Transport, error)
  • func (t *Transport) ChannelConnectivityStateForTesting() connectivity.State
  • func (t *Transport) Close()
  • func (t *Transport) ReportLoad() (*load.Store, func())
  • func (t *Transport) SendRequest(url string, resources []string)

Types

type ADSFlowControl

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

ADSFlowControl implements ADS stream level flow control that enables the transport to block the reading of the next message off of the stream until the previous update is consumed by all watchers.

The lifetime of the flow control is tied to the lifetime of the stream.

New instances must be created with a call to NewADSStreamFlowControl.

func NewADSStreamFlowControl

func NewADSStreamFlowControl() *ADSFlowControl

NewADSStreamFlowControl returns a new ADSFlowControl.

func (*ADSFlowControl) Add

func (fc *ADSFlowControl) Add()

Add increments the number of watchers (by one) who are yet to consume the most recent update received on the ADS stream.

func (*ADSFlowControl) OnDone

func (fc *ADSFlowControl) OnDone()

OnDone indicates that a watcher has consumed the most recent update.

func (*ADSFlowControl) Wait

func (fc *ADSFlowControl) Wait(ctx context.Context) bool

Wait blocks until all the watchers have consumed the most recent update and returns true. If the context expires before that, it returns false.

type OnRecvHandlerFunc

type OnRecvHandlerFunc func(update ResourceUpdate, fc *ADSFlowControl) error

OnRecvHandlerFunc is the implementation at the xDS data model layer, which determines if the configuration received from the management server can be applied locally or not.

A nil error is returned from this function when the data model layer believes that the received configuration is good and can be applied locally. This will cause the transport layer to send an ACK to the management server. A non-nil error is returned from this function when the data model layer believes otherwise, and this will cause the transport layer to send a NACK.

The implementation is expected to use the ADS flow control object passed to it, and increment the number of watchers to whom the update is sent to, and eventually decrement the number once the update is consumed by the watchers.

type OnSendHandlerFunc

type OnSendHandlerFunc func(update *ResourceSendInfo)

OnSendHandlerFunc is the implementation at the authority, which handles state changes for the resource watch and stop watch timers accordingly.

type Options

type Options struct {
	// ServerCfg contains all the configuration required to connect to the xDS
	// management server.
	ServerCfg *bootstrap.ServerConfig
	// OnRecvHandler is the component which makes ACK/NACK decisions based on
	// the received resources.
	//
	// Invoked inline and implementations must not block.
	OnRecvHandler OnRecvHandlerFunc
	// OnErrorHandler provides a way for the transport layer to report
	// underlying stream errors. These can be bubbled all the way up to the user
	// of the xdsClient.
	//
	// Invoked inline and implementations must not block.
	OnErrorHandler func(error)
	// OnSendHandler provides a way for the transport layer to report underlying
	// resource requests sent on the stream. However, Send() on the ADS stream will
	// return successfully as long as:
	//   1. there is enough flow control quota to send the message.
	//   2. the message is added to the send buffer.
	// However, the connection may fail after the callback is invoked and before
	// the message is actually sent on the wire. This is accepted.
	//
	// Invoked inline and implementations must not block.
	OnSendHandler func(*ResourceSendInfo)
	// Backoff controls the amount of time to backoff before recreating failed
	// ADS streams. If unspecified, a default exponential backoff implementation
	// is used. For more details, see:
	// https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md.
	Backoff func(retries int) time.Duration
	// Logger does logging with a prefix.
	Logger *grpclog.PrefixLogger
	// NodeProto contains the Node proto to be used in xDS requests. This will be
	// of type *v3corepb.Node.
	NodeProto *v3corepb.Node
}

Options specifies configuration knobs used when creating a new Transport.

type ResourceSendInfo

type ResourceSendInfo struct {
	ResourceNames []string
	URL           string
}

ResourceSendInfo wraps the names and url of resources sent to the management server. This is used by the `authority` type to start/stop the watch timer associated with every resource in the update.

type ResourceUpdate

type ResourceUpdate struct {
	// Resources is the list of resources received from the management server.
	Resources []*anypb.Any
	// URL is the resource type URL for the above resources.
	URL string
	// Version is the resource version, for the above resources, as specified by
	// the management server.
	Version string
}

ResourceUpdate is a representation of the configuration update received from the management server. It only contains fields which are useful to the data model layer, and layers above it.

type Transport

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

Transport provides a resource-type agnostic implementation of the xDS transport protocol. At this layer, resource contents are supposed to be opaque blobs which should be meaningful only to the xDS data model layer which is implemented by the `xdsresource` package.

Under the hood, it owns the gRPC connection to a single management server and manages the lifecycle of ADS/LRS streams. It uses the xDS v3 transport protocol version.

func New

func New(opts Options) (*Transport, error)

New creates a new Transport.

func (*Transport) ChannelConnectivityStateForTesting

func (t *Transport) ChannelConnectivityStateForTesting() connectivity.State

ChannelConnectivityStateForTesting returns the connectivity state of the gRPC channel to the management server.

Only for testing purposes.

func (*Transport) Close

func (t *Transport) Close()

Close closes the Transport and frees any associated resources.

func (*Transport) ReportLoad

func (t *Transport) ReportLoad() (*load.Store, func())

ReportLoad starts reporting loads to the management server the transport is configured to use.

It returns a Store for the user to report loads and a function to cancel the load reporting.

func (*Transport) SendRequest

func (t *Transport) SendRequest(url string, resources []string)

SendRequest sends out an ADS request for the provided resources of the specified resource type.

The request is sent out asynchronously. If no valid stream exists at the time of processing this request, it is queued and will be sent out once a valid stream exists.

If a successful response is received, the update handler callback provided at creation time is invoked. If an error is encountered, the stream error handler callback provided at creation time is invoked.