xds

Documentation

Index

• type EndpointAddress
• type Features
• type FixedHash
  • func (FixedHash) ID(_ *corev3.Node) string
• type GRPCServerListenerCache
  • func NewGRPCServerListenerCache() *GRPCServerListenerCache
  • func (c *GRPCServerListenerCache) Add(nodeHash string, newAddresses []EndpointAddress) bool
  • func (c *GRPCServerListenerCache) Get(nodeHash string) []EndpointAddress
• type SnapshotBuilder
  • func NewSnapshotBuilder(nodeHash string, localityPriorityMapper eds.LocalityPriorityMapper, ...) *SnapshotBuilder
  • func (b *SnapshotBuilder) AddGRPCApplications(apps []applications.Application) (*SnapshotBuilder, error)
  • func (b *SnapshotBuilder) AddGRPCServerListenerAddresses(addresses []EndpointAddress) *SnapshotBuilder
  • func (b *SnapshotBuilder) Build() (cachev3.ResourceSnapshot, error)
• type SnapshotCache
  • func NewSnapshotCache(ctx context.Context, allowPartialRequests bool, hash cachev3.NodeHash, ...) *SnapshotCache
  • func (c *SnapshotCache) CreateDeltaWatch(request *cachev3.DeltaRequest, state streamv3.StreamState, ...) (cancel func())
  • func (c *SnapshotCache) CreateWatch(request *cachev3.Request, state streamv3.StreamState, ...) (cancel func())
  • func (c *SnapshotCache) Fetch(ctx context.Context, request *cachev3.Request) (cachev3.Response, error)
  • func (c *SnapshotCache) UpdateResources(_ context.Context, logger logr.Logger, kubecontextName string, ...) error
• type ZoneHash
  • func (ZoneHash) ID(node *corev3.Node) string

Types

type EndpointAddress

type EndpointAddress struct {
	Host string
	Port uint32
}

EndpointAddress represents a socket ipAddress, with an IP address (e.g., "0.0.0.0" or "[::]"), and a port.

type Features

type Features struct {
	EnableControlPlaneTLS          bool `yaml:"enableControlPlaneTls"`
	RequireControlPlaneClientCerts bool `yaml:"requireControlPlaneClientCerts"`
	EnableDataPlaneTLS             bool `yaml:"enableDataPlaneTls"`
	RequireDataPlaneClientCerts    bool `yaml:"requireDataPlaneClientCerts"`
	EnableRBAC                     bool `yaml:"enableRbac"`
	EnableFederation               bool `yaml:"enableFederation"`
}

Features of the xDS control plane that can be enabled and disabled via a config file.

type FixedHash

type FixedHash struct{}

FixedHash uses a fixed value as the node hash.

func (FixedHash) ID

func (FixedHash) ID(_ *corev3.Node) string

type GRPCServerListenerCache

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

func NewGRPCServerListenerCache

func NewGRPCServerListenerCache() *GRPCServerListenerCache

func (*GRPCServerListenerCache) Add

func (c *GRPCServerListenerCache) Add(nodeHash string, newAddresses []EndpointAddress) bool

Add returns true if at least one of the new gRPC server listener addresses did not already exist in the cache for the provided `nodeHash` cache key.

func (*GRPCServerListenerCache) Get

func (c *GRPCServerListenerCache) Get(nodeHash string) []EndpointAddress

type SnapshotBuilder

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

SnapshotBuilder builds xDS resource snapshots for the cache.

func NewSnapshotBuilder

func NewSnapshotBuilder(nodeHash string, localityPriorityMapper eds.LocalityPriorityMapper, features *Features, authority string) *SnapshotBuilder

NewSnapshotBuilder initializes the builder.

func (*SnapshotBuilder) AddGRPCApplications

func (b *SnapshotBuilder) AddGRPCApplications(apps []applications.Application) (*SnapshotBuilder, error)

AddGRPCApplications adds the provided application configurations to the xDS resource snapshot.

func (*SnapshotBuilder) AddGRPCServerListenerAddresses

func (b *SnapshotBuilder) AddGRPCServerListenerAddresses(addresses []EndpointAddress) *SnapshotBuilder

AddGRPCServerListenerAddresses adds server listeners and associated route configurations with the provided IP addresses and ports to the snapshot.

func (*SnapshotBuilder) Build

func (b *SnapshotBuilder) Build() (cachev3.ResourceSnapshot, error)

Build adds the server listeners and route configuration for the node hash, and then builds the snapshot.

type SnapshotCache

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

SnapshotCache stores snapshots of xDS resources in a delegate cache.

It handles server listener requests by intercepting Listener stream creation, see `CreateWatch()`. Server listeners addresses from these requests are kept in a map, keyed by the node hash, and with a set of addresses per node hash.

It also handles propagating snapshots to all node hashes in the cache.

func NewSnapshotCache

func NewSnapshotCache(ctx context.Context, allowPartialRequests bool, hash cachev3.NodeHash, localityPriorityMapper eds.LocalityPriorityMapper, features *Features, authority string) *SnapshotCache

NewSnapshotCache creates an xDS resource cache for the provided node hash function.

If `allowPartialRequests` is true, the DiscoveryServer will respond to requests for a resource type even if some resources in the snapshot are not named in the request.

func (*SnapshotCache) CreateDeltaWatch

func (c *SnapshotCache) CreateDeltaWatch(request *cachev3.DeltaRequest, state streamv3.StreamState, responses chan cachev3.DeltaResponse) (cancel func())

CreateDeltaWatch just delegates, since gRPC does not support delta/incremental xDS currently. TODO: Handle request for gRPC server Listeners once gRPC implementation support delta/incremental xDS.

func (*SnapshotCache) CreateWatch

func (c *SnapshotCache) CreateWatch(request *cachev3.Request, state streamv3.StreamState, responses chan cachev3.Response) (cancel func())

CreateWatch intercepts stream creation before delegating, and if it is a request for Listener (LDS) resources stream, does the following:

• Extracts addresses and ports of any server listeners in the request and adds them to the set of known server listener socket addresses for the node hash.
• If there is no existing snapshot, or if the request contained new and previously unseen server listener addresses the node hash, creates a new snapshot for that node hash, with the server listeners and associated route configuration.

This solves bootstrapping of xDS resources snapshots for xDS-enabled gRPC servers and Envoy proxy instances that fetch configuration dynamically using ADS.

func (*SnapshotCache) Fetch

func (c *SnapshotCache) Fetch(ctx context.Context, request *cachev3.Request) (cachev3.Response, error)

func (*SnapshotCache) UpdateResources

func (c *SnapshotCache) UpdateResources(_ context.Context, logger logr.Logger, kubecontextName string, namespace string, updatedApps []applications.Application) error

UpdateResources creates a new snapshot for each node hash in the cache, based on the provided gRPC application configuration, with the addition of server listeners and their associated route configurations.

type ZoneHash

type ZoneHash struct{}

ZoneHash uses `locality.zone` as the node hash, so all xDS clients in the same zone access the same cache snapshot.

func (ZoneHash) ID

func (ZoneHash) ID(node *corev3.Node) string