ads

Documentation

Overview

Package ads implements Envoy's Aggregated Discovery Service (ADS).

Index

• Constants
• func GetProxyFromPod(pod *v1.Pod) (*envoy.Proxy, error)
• type Callbacks
  • func (cb *Callbacks) OnDeltaStreamClosed(id int64)
  • func (cb *Callbacks) OnDeltaStreamOpen(_ context.Context, id int64, typ string) error
  • func (cb *Callbacks) OnFetchRequest(_ context.Context, req *discovery.DiscoveryRequest) error
  • func (cb *Callbacks) OnFetchResponse(req *discovery.DiscoveryRequest, resp *discovery.DiscoveryResponse)
  • func (cb *Callbacks) OnStreamClosed(id int64)
  • func (cb *Callbacks) OnStreamDeltaRequest(a int64, req *discovery.DeltaDiscoveryRequest) error
  • func (cb *Callbacks) OnStreamDeltaResponse(a int64, req *discovery.DeltaDiscoveryRequest, ...)
  • func (cb *Callbacks) OnStreamOpen(_ context.Context, id int64, typ string) error
  • func (cb *Callbacks) OnStreamRequest(a int64, req *discovery.DiscoveryRequest) error
  • func (cb *Callbacks) OnStreamResponse(_ context.Context, aa int64, req *discovery.DiscoveryRequest, ...)
• type Server
  • func NewADSServer(meshCatalog catalog.MeshCataloger, proxyRegistry *registry.ProxyRegistry, ...) *Server
  • func (s *Server) DeltaAggregatedResources(xds_discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error
  • func (s *Server) GetID() string
  • func (s *Server) GetXDSLog() map[string]map[envoy.TypeURI][]time.Time
  • func (s *Server) Liveness() bool
  • func (s *Server) Readiness() bool
  • func (s *Server) RecordFullSnapshot(proxy *envoy.Proxy, snapshotResources map[string][]types.Resource) error
  • func (s *Server) SendDiscoveryResponse(proxy *envoy.Proxy, request *xds_discovery.DiscoveryRequest, ...) error
  • func (s *Server) Start(ctx context.Context, cancel context.CancelFunc, port int, ...) error
  • func (s *Server) StreamAggregatedResources(...) error

Constants

const (
	// MaxXdsLogsPerProxy keeps a higher bound of how many timestamps do we keep per proxy
	MaxXdsLogsPerProxy = 20
)

const (
	// ServerType is the type identifier for the ADS server
	ServerType = "ADS"
)

Functions

func GetProxyFromPod

func GetProxyFromPod(pod *v1.Pod) (*envoy.Proxy, error)

GetProxyFromPod infers and creates a Proxy data structure from a Pod. This is a temporary workaround as proxy is required and expected in any vertical call to XDS, however snapshotcache has no need to provide visibility on proxies whatsoever. All verticals use the proxy structure to infer the pod later, so the actual only mandatory data for the verticals to be functional is the common name, which links proxy <-> pod

Types

type Callbacks

type Callbacks struct {
}

Callbacks is an implementation of xDS server callbacks required by go-control-plane These are put in case we want to add any logic to specific parts of the xDS server proto handling implementation. Though mandatory to be provided, they are not required to do anything, but can certainly help to understand, debug and instrument additional functionality on top of the cache. Sample implementation from https://github.com/envoyproxy/go-control-plane/blob/main/docs/cache/Server.md

func (*Callbacks) OnDeltaStreamClosed

func (cb *Callbacks) OnDeltaStreamClosed(id int64)

OnDeltaStreamClosed is called when a Delta stream is being closed

func (*Callbacks) OnDeltaStreamOpen

func (cb *Callbacks) OnDeltaStreamOpen(_ context.Context, id int64, typ string) error

OnDeltaStreamOpen is called when a Delta stream is being opened

func (*Callbacks) OnFetchRequest

func (cb *Callbacks) OnFetchRequest(_ context.Context, req *discovery.DiscoveryRequest) error

OnFetchRequest is called when a fetch request is received

func (*Callbacks) OnFetchResponse

func (cb *Callbacks) OnFetchResponse(req *discovery.DiscoveryRequest, resp *discovery.DiscoveryResponse)

OnFetchResponse is called when a fetch request is being responded to

func (*Callbacks) OnStreamClosed

func (cb *Callbacks) OnStreamClosed(id int64)

OnStreamClosed is called on stream closed

func (*Callbacks) OnStreamDeltaRequest

func (cb *Callbacks) OnStreamDeltaRequest(a int64, req *discovery.DeltaDiscoveryRequest) error

OnStreamDeltaRequest is called when a Delta request comes on an open Delta stream

func (*Callbacks) OnStreamDeltaResponse

func (cb *Callbacks) OnStreamDeltaResponse(a int64, req *discovery.DeltaDiscoveryRequest, resp *discovery.DeltaDiscoveryResponse)

OnStreamDeltaResponse is called when a Delta request is getting responded to

func (*Callbacks) OnStreamOpen

func (cb *Callbacks) OnStreamOpen(_ context.Context, id int64, typ string) error

OnStreamOpen is called on stream open

func (*Callbacks) OnStreamRequest

func (cb *Callbacks) OnStreamRequest(a int64, req *discovery.DiscoveryRequest) error

OnStreamRequest is called when a request happens on an open string

func (*Callbacks) OnStreamResponse

func (cb *Callbacks) OnStreamResponse(_ context.Context, aa int64, req *discovery.DiscoveryRequest, resp *discovery.DiscoveryResponse)

OnStreamResponse is called when a response is being sent to a request

type Server

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

Server implements the Envoy xDS Aggregate Discovery Services

func NewADSServer

func NewADSServer(meshCatalog catalog.MeshCataloger, proxyRegistry *registry.ProxyRegistry, enableDebug bool, osmNamespace string,
	cfg configurator.Configurator, certManager *certificate.Manager, kubecontroller k8s.Controller, msgBroker *messaging.Broker) *Server

NewADSServer creates a new Aggregated Discovery Service server

func (*Server) DeltaAggregatedResources

func (s *Server) DeltaAggregatedResources(xds_discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error

DeltaAggregatedResources implements discovery.AggregatedDiscoveryServiceServer

func (*Server) GetID

func (s *Server) GetID() string

GetID returns the ID of the probe

func (*Server) GetXDSLog

func (s *Server) GetXDSLog() map[string]map[envoy.TypeURI][]time.Time

GetXDSLog implements XDSDebugger interface and a log of the XDS responses sent to Envoy proxies.

func (*Server) Liveness

func (s *Server) Liveness() bool

Liveness is the Kubernetes liveness probe handler.

func (*Server) Readiness

func (s *Server) Readiness() bool

Readiness is the Kubernetes readiness probe handler.

func (*Server) RecordFullSnapshot

func (s *Server) RecordFullSnapshot(proxy *envoy.Proxy, snapshotResources map[string][]types.Resource) error

RecordFullSnapshot stores a group of resources as a new Snapshot with a new version in the cache. It also runs a consistency check on the snapshot (will warn if there are missing resources referenced in the snapshot)

func (*Server) SendDiscoveryResponse

func (s *Server) SendDiscoveryResponse(proxy *envoy.Proxy, request *xds_discovery.DiscoveryRequest, server *xds_discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer, resourcesToSend []types.Resource) error

SendDiscoveryResponse creates a new response for <proxy> given <resourcesToSend> and <request.TypeURI> and sends it

func (*Server) Start

func (s *Server) Start(ctx context.Context, cancel context.CancelFunc, port int, adsCert *certificate.Certificate) error

Start starts the ADS server

func (*Server) StreamAggregatedResources

func (s *Server) StreamAggregatedResources(server xds_discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error

StreamAggregatedResources handles streaming of the clusters to the connected Envoy proxies This is evaluated once per new Envoy proxy connecting and remains running for the duration of the gRPC socket.