krtxds

package
v1.4.0 Latest Latest
Warning

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

 Go to latest Published: Jul 27, 2026 License: Apache-2.0 Imports: 36 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	TargetTypeResourceUrl = "type.googleapis.com/agentgateway.dev.resource.Resource"
	TargetTypeAddressUrl  = "type.googleapis.com/istio.workload.Address"
)
View Source 
const (
	// KeyDelimiter is the character used to join segments of a cache key
	KeyDelimiter = "~"

	// RoleKey is the name of the ket in the node.metadata used to store the role
	RoleKey = "role"

	// PeerCtxKey is the key used to store the peer information in the context
	PeerCtxKey = "peer"
)

Variables

View Source 
var (
	DebounceAfter = env.Register(
		"KGW_DEBOUNCE_AFTER",
		10*time.Millisecond,
		"The delay added to config/registry events for debouncing. This will delay the push by "+
			"at least this interval. If no change is detected within this period, the push will happen, "+
			" otherwise we'll keep delaying until things settle, up to a max of KGW_DEBOUNCE_MAX.",
	).Get()

	DebounceMax = env.Register(
		"KGW_DEBOUNCE_MAX",
		1*time.Second,
		"The maximum amount of time to wait for events while debouncing. If events keep showing up with no breaks "+
			"for this time, we'll trigger a push.",
	).Get()
)

Functions

func AgentgatewayID 

func AgentgatewayID(node *envoycorev3.Node) types.NamespacedName

func ByteCount 

func ByteCount(b int) string

ByteCount returns a human readable byte format Inspired by https://yourbasic.org/golang/formatting-byte-size-to-human-readable-format/

func GetShortType 

func GetShortType(typeURL string) string

GetShortType returns an abbreviated form of a type, useful for logging or human friendly messages

func ResourceSize 

func ResourceSize(r model.Resources) int

func TypeName 

func TypeName[T proto.Message]() string

Types

type CollectionGenerator 

type CollectionGenerator struct {
	PerGateway bool
	Col        krt.Collection[DiscoveryResource]
}

func (CollectionGenerator) GenerateDeltas 

func (e CollectionGenerator) GenerateDeltas(req *PushRequest, w *model.WatchedResource, gw types.NamespacedName) (model.Resources, model.DeletedResources, error)

GenerateDeltas computes Workload resources. This is design to be highly optimized to delta updates, and supports *on-demand* client usage. A client can subscribe with a wildcard subscription and get all resources (with delta updates), or on-demand and only get responses for specifically subscribed resources.

Incoming requests may be for VIP or Pod IP addresses. However, all responses are Workload resources, which are pod based. This means subscribing to a VIP may end up pushing many resources of different name than the request. On-demand clients are expected to handle this (for wildcard, this is not applicable, as they don't specify any resources at all).

type CollectionRegistration 

type CollectionRegistration struct {
	Start     func(stop <-chan struct{})
	HasSynced func() bool
}

type Connection 

type Connection struct {
	xds.Connection
	// contains filtered or unexported fields
}

type DebounceOptions 

type DebounceOptions struct {
	// DebounceAfter is the delay added to events to wait
	// after a registry/config event for debouncing.
	// This will delay the push by at least this interval, plus
	// the time getting subsequent events. If no change is
	// detected the push will happen, otherwise we'll keep
	// delaying until things settle.
	DebounceAfter time.Duration

	// debounceMax is the maximum time to wait for events
	// while debouncing. Defaults to 10 seconds. If events keep
	// showing up with no break for this time, we'll trigger a push.
	DebounceMax time.Duration
}

type DeltaDiscoveryStream 

type DeltaDiscoveryStream = discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesServer

type DiscoveryResource 

type DiscoveryResource struct {
	*discovery.Resource
	ForGateway *types.NamespacedName
}

func (DiscoveryResource) Equals 

func (d DiscoveryResource) Equals(other DiscoveryResource) bool

func (DiscoveryResource) IsForGateway 

func (d DiscoveryResource) IsForGateway(other types.NamespacedName) bool

func (DiscoveryResource) ResourceName 

func (d DiscoveryResource) ResourceName() string

type DiscoveryServer 

type DiscoveryServer struct {
	// Generators allow customizing the generated config, based on the client metadata.
	// Key is the generator type - will match the Generator metadata to set the per-connection
	// default generator, or the combination of Generator metadata and TypeUrl to select a
	// different generator for a type.
	// Normal istio clients use the default generator - will not be impacted by this.
	Collections map[string]CollectionGenerator

	// RequestRateLimit limits the number of new XDS requests allowed. This helps prevent thundering hurd of incoming requests.
	RequestRateLimit *rate.Limiter

	// InboundUpdates describes the number of configuration updates the discovery server has received
	InboundUpdates *atomic.Int64
	// CommittedUpdates describes the number of configuration updates the discovery server has
	// received, process, and stored in the push context. If this number is less than InboundUpdates,
	// there are updates we have not yet processed.
	// Note: This does not mean that all proxies have received these configurations; it is strictly
	// the push context, which means that the next push to a proxy will receive this configuration.
	CommittedUpdates *atomic.Int64

	DebounceOptions DebounceOptions
	// contains filtered or unexported fields
}

DiscoveryServer is Pilot's gRPC implementation for Envoy's xds APIs

func NewDiscoveryServer 

func NewDiscoveryServer(debugger *krt.DebugHandler, nackPublisher *nack.Publisher, reg ...Registration) *DiscoveryServer

NewDiscoveryServer creates a DiscoveryServer for agentgateway that sources data from KRT collections via registered generators

func (*DiscoveryServer) AllClients 

func (s *DiscoveryServer) AllClients() []*Connection

AllClients returns all connected clients, per Clients, but additionally includes uninitialized connections Warning: callers must take care not to rely on the con.proxy field being set

func (*DiscoveryServer) Clients 

func (s *DiscoveryServer) Clients() []*Connection

Clients returns all currently connected clients. This method can be safely called concurrently, but care should be taken with the underlying objects (ie model.Proxy) to ensure proper locking. This method returns only fully initialized connections; for all connections, use AllClients

func (*DiscoveryServer) DeltaAggregatedResources 

func (s *DiscoveryServer) DeltaAggregatedResources(stream discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error

func (*DiscoveryServer) EnsureSynced 

func (s *DiscoveryServer) EnsureSynced()

EnsureSynced waits until all pending debounce events have been processed. This is useful in tests to ensure that no spurious pushes will occur after connecting a client.

func (*DiscoveryServer) IsServerReady 

func (s *DiscoveryServer) IsServerReady() bool

func (*DiscoveryServer) NextVersion 

func (s *DiscoveryServer) NextVersion() string

func (*DiscoveryServer) ProxyNeedsPush 

func (s *DiscoveryServer) ProxyNeedsPush(proxy *Proxy, request *PushRequest) bool

func (*DiscoveryServer) Push 

func (s *DiscoveryServer) Push(req *PushRequest)

Push is called to push changes on config updates using ADS.

func (*DiscoveryServer) Shutdown 

func (s *DiscoveryServer) Shutdown()

Shutdown shuts down DiscoveryServer components.

func (*DiscoveryServer) Start 

func (s *DiscoveryServer) Start(stopCh <-chan struct{})

func (*DiscoveryServer) StreamAggregatedResources 

func (s *DiscoveryServer) StreamAggregatedResources(stream discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error

StreamAggregatedResources implements the ADS interface.

func (*DiscoveryServer) StreamDeltas 

func (s *DiscoveryServer) StreamDeltas(stream DeltaDiscoveryStream) error

func (*DiscoveryServer) WaitForRequestLimit 

func (s *DiscoveryServer) WaitForRequestLimit(ctx context.Context) error

type Event 

type Event struct {
	// PushRequest PushRequest to use for the push.
	PushRequest *PushRequest

	// function to call once a push is finished. This must be called or future changes may be blocked.
	Done func()
}

Event represents a config or registry event that results in a push.

type IntoProto 

type IntoProto[T proto.Message] interface {
	IntoProto() T
}

type IntoResourceName 

type IntoResourceName interface {
	XDSResourceName() string
}

type Proxy 

type Proxy struct {
	sync.RWMutex

	// ID is the unique identifier for a client.
	ID string

	// WatchedResources contains the list of watched resources for the proxy, keyed by the DiscoveryRequest TypeUrl.
	WatchedResources map[string]*model.WatchedResource
}

Proxy contains information about an specific instance of a proxy. The Proxy is initialized when a client connects to XDS, and populated from 'node' info in the protocol as well as data extracted from registries.

In current implementation nodes use a 4-parts '~' delimited ID. Type~IPAddress~ID~Domain

func (*Proxy) AddOrUpdateWatchedResource 

func (node *Proxy) AddOrUpdateWatchedResource(r *model.WatchedResource)

func (*Proxy) DeepCloneWatchedResources 

func (node *Proxy) DeepCloneWatchedResources() map[string]model.WatchedResource

DeepCloneWatchedResources clones the watched resources

func (*Proxy) DeleteWatchedResource 

func (node *Proxy) DeleteWatchedResource(typeURL string)

func (*Proxy) GetWatchedResource 

func (node *Proxy) GetWatchedResource(typeURL string) *model.WatchedResource

func (*Proxy) GetWatchedResourceTypes 

func (node *Proxy) GetWatchedResourceTypes() sets.String

func (*Proxy) ShallowCloneWatchedResources 

func (node *Proxy) ShallowCloneWatchedResources() map[string]*model.WatchedResource

ShallowCloneWatchedResources clones the watched resources, both the keys and values are shallow copy.

func (*Proxy) UpdateWatchedResource 

func (node *Proxy) UpdateWatchedResource(typeURL string, updateFn func(*model.WatchedResource) *model.WatchedResource)

type PushQueue 

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

func NewPushQueue 

func NewPushQueue() *PushQueue

func (*PushQueue) Dequeue 

func (p *PushQueue) Dequeue() (con *Connection, request *PushRequest, shutdown bool)

Remove a proxy from the queue. If there are no proxies ready to be removed, this will block

func (*PushQueue) Enqueue 

func (p *PushQueue) Enqueue(con *Connection, pushRequest *PushRequest)

Enqueue will mark a proxy as pending a push. If it is already pending, pushInfo will be merged. ServiceEntry updates will be added together, and full will be set if either were full

func (*PushQueue) MarkDone 

func (p *PushQueue) MarkDone(con *Connection)

func (*PushQueue) Pending 

func (p *PushQueue) Pending() int

Get number of pending proxies

func (*PushQueue) ShutDown 

func (p *PushQueue) ShutDown()

ShutDown will cause queue to ignore all new items added to it. As soon as the worker goroutines have drained the existing items in the queue, they will be instructed to exit.

type PushRequest 

type PushRequest struct {
	// ConfigsUpdated keeps track of configs that have changed.
	ConfigsUpdated map[TypeUrl]sets.String

	IsFromRequest bool
	// InitialResourceVersions contains the versions the client retained across a reconnect.
	InitialResourceVersions map[string]string

	// PushVersion represent the version of the push
	PushVersion string

	// Delta defines the resources that were added or removed as part of this push request.
	// This is set only on requests from the client which change the set of resources they (un)subscribe from.
	Delta xds.ResourceDelta
}

PushRequest defines a request to push to proxies It is used to send updates to the config update debouncer and pass to the PushQueue.

func (*PushRequest) CopyMerge added in v1.3.0 

func (pr *PushRequest) CopyMerge(other *PushRequest) *PushRequest

CopyMerge merges two update requests without mutating either input. This should be used after a PushRequest has been fanned out and may be shared by multiple proxy queues or push goroutines.

func (PushRequest) IsRequest 

func (r PushRequest) IsRequest() bool

func (*PushRequest) Merge 

func (pr *PushRequest) Merge(other *PushRequest) *PushRequest

Merge two update requests together Merge behaves similarly to a list append; usage should in the form `a = a.merge(b)`. Importantly, Merge may decide to allocate a new PushRequest object or reuse the existing one - both inputs should not be used after completion.

func (*PushRequest) PushReason 

func (pr *PushRequest) PushReason() string

type Registration 

type Registration func(*DiscoveryServer) CollectionRegistration

func Collection 

func Collection[T IntoProto[TT], TT proto.Message](collection krt.Collection[T], krtopts krtutil.KrtOptions) Registration

func PerGatewayCollection 

func PerGatewayCollection[T IntoProto[TT], TT proto.Message](collection krt.Collection[T], extract func(o T) types.NamespacedName, krtopts krtutil.KrtOptions) Registration

type TypeUrl 

type TypeUrl string

Source Files

View all Source files

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.