sink

Documentation

Index

• type AuthChecker
• type Change
• type Client
  • func NewClient(client mcp.ResourceSourceClient, options *Options) *Client
  • func (c *Client) Run(ctx context.Context)
• type CollectionOptions
  • func CollectionOptionsFromSlice(names []string) []CollectionOptions
• type InMemoryUpdater
  • func NewInMemoryUpdater() *InMemoryUpdater
  • func (u *InMemoryUpdater) Apply(c *Change) error
  • func (u *InMemoryUpdater) Get(collection string) []*Object
• type JournaledRequest
  • func (jr *JournaledRequest) ToMeshConfigRequest() *mcp.MeshConfigRequest
• type Object
• type Options
• type RecentRequestInfo
  • func (r RecentRequestInfo) Acked() bool
• type RecentRequestsJournal
  • func NewRequestJournal() *RecentRequestsJournal
  • func (r *RecentRequestsJournal) RecordMeshConfigRequest(req *mcp.MeshConfigRequest)
  • func (r *RecentRequestsJournal) RecordRequestResources(req *mcp.RequestResources)
  • func (r *RecentRequestsJournal) Snapshot() []RecentRequestInfo
• type Server
  • func NewServer(srcOptions *Options, serverOptions *ServerOptions) *Server
  • func (s *Server) EstablishResourceStream(stream mcp.ResourceSink_EstablishResourceStreamServer) error
• type ServerOptions
• type Sink
  • func New(options *Options) *Sink
  • func (sink *Sink) Collections() []string
  • func (sink *Sink) ID() string
  • func (sink *Sink) Metadata() map[string]string
  • func (sink *Sink) SnapshotRequestInfo() []RecentRequestInfo
• type Stream
• type Updater

Types

type AuthChecker

type AuthChecker interface {
	Check(authInfo credentials.AuthInfo) error
}

AuthChecker is used to check the transport auth info that is associated with each stream. If the function returns nil, then the connection will be allowed. If the function returns an error, then it will be percolated up to the gRPC stack.

Note that it is possible that this method can be called with nil authInfo. This can happen either if there is no peer info, or if the underlying gRPC stream is insecure. The implementations should be resilient in this case and apply appropriate policy.

type Change

type Change struct {
	Collection string

	// List of resources to add/update. The interpretation of this field depends
	// on the value of Incremental.
	//
	// When Incremental=True, the list only includes new/updateReceivedForStream resources.
	//
	// When Incremental=False, the list includes the full list of resources.
	// Any previously received resources not in this list should be deleted.
	Objects []*Object

	// List of deleted resources by name. The resource name corresponds to the
	// resource's metadata name.
	//
	// Ignore when Incremental=false.
	Removed []string

	// When true, the set of changes represents an requestIncremental resource update. The
	// `Objects` is a list of added/update resources and `Removed` is a list of delete
	// resources.
	//
	// When false, the set of changes represents a full-state update for the specified
	// type. Any previous resources not included in this update should be removed.
	Incremental bool

	// SystemVersionInfo is the version of the response data (used for debugging purposes only).
	SystemVersionInfo string
}

changes is a collection of configuration objects of the same protobuf type.

type Client

type Client struct {
	*Sink
	// contains filtered or unexported fields
}

Client implements the client for the MCP source service. The client is the sink and receives configuration from the server.

func NewClient

func NewClient(client mcp.ResourceSourceClient, options *Options) *Client

func (*Client) Run

func (c *Client) Run(ctx context.Context)

type CollectionOptions

type CollectionOptions struct {
	// Name of the collection, e.g. istio/networking/v1alpha3/VirtualService
	Name string

	// When true, the sink requests incremental updates from the source. Incremental
	// updates are requested when this option is true. Incremental updates are only
	// used if the sink requests it (per request) and the source decides to make use of it.
	Incremental bool
}

CollectionOptions configures the per-collection updates.

func CollectionOptionsFromSlice

func CollectionOptionsFromSlice(names []string) []CollectionOptions

CollectionOptionsFromSlice returns a slice of collection options from a slice of collection names.

type InMemoryUpdater

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

InMemoryUpdater is an implementation of Updater that keeps a simple in-memory state.

func NewInMemoryUpdater

func NewInMemoryUpdater() *InMemoryUpdater

NewInMemoryUpdater returns a new instance of InMemoryUpdater

func (*InMemoryUpdater) Apply

func (u *InMemoryUpdater) Apply(c *Change) error

Apply the change to the InMemoryUpdater.

func (*InMemoryUpdater) Get

func (u *InMemoryUpdater) Get(collection string) []*Object

Get current state for the given collection.

type JournaledRequest

type JournaledRequest struct {
	VersionInfo   string
	Collection    string
	ResponseNonce string
	ErrorDetail   *rpc.Status
	SinkNode      *mcp.SinkNode
}

JournaledRequest is a common structure for journaling both mcp.MeshConfigRequest and mcp.RequestResources. It can be replaced with mcp.RequestResources once we fully switch over to the new API.

func (*JournaledRequest) ToMeshConfigRequest

func (jr *JournaledRequest) ToMeshConfigRequest() *mcp.MeshConfigRequest

type Object

type Object struct {
	TypeURL  string
	Metadata *mcp.Metadata
	Body     proto.Message
}

Object contains a decoded versioned object with metadata received from the server.

type Options

type Options struct {
	CollectionOptions []CollectionOptions
	Updater           Updater
	ID                string
	Metadata          map[string]string
	Reporter          monitoring.Reporter
}

Options contains options for configuring MCP sinks.

type RecentRequestInfo

type RecentRequestInfo struct {
	Time    time.Time
	Request *JournaledRequest
}

RecentRequestInfo is metadata about a request that the client has sent.

func (RecentRequestInfo) Acked

func (r RecentRequestInfo) Acked() bool

Acked indicates whether the message was an ack or not.

type RecentRequestsJournal

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

RecentRequestsJournal captures debug metadata about the latest requests that was sent by this client.

func NewRequestJournal

func NewRequestJournal() *RecentRequestsJournal

func (*RecentRequestsJournal) RecordMeshConfigRequest

func (r *RecentRequestsJournal) RecordMeshConfigRequest(req *mcp.MeshConfigRequest)

func (*RecentRequestsJournal) RecordRequestResources

func (r *RecentRequestsJournal) RecordRequestResources(req *mcp.RequestResources)

func (*RecentRequestsJournal) Snapshot

func (r *RecentRequestsJournal) Snapshot() []RecentRequestInfo

type Server

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

Server implements the server for the MCP sink service. The server is the sink and receives configuration from the client.

func NewServer

func NewServer(srcOptions *Options, serverOptions *ServerOptions) *Server

NewServer creates a new instance of a MCP sink server.

func (*Server) EstablishResourceStream

func (s *Server) EstablishResourceStream(stream mcp.ResourceSink_EstablishResourceStreamServer) error

EstablishResourceStream implements the ResourceSinkServer interface.

type ServerOptions

type ServerOptions struct {
	NewConnectionFreq      time.Duration
	NewConnectionBurstSize int
	AuthChecker            AuthChecker
}

ServerOptions contains source server specific options

type Sink

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

Sink implements the resource sink message exchange for MCP. It can be instantiated by client and server sink implementations to manage the MCP message exchange.

func New

func New(options *Options) *Sink

New creates a new resource sink.

func (*Sink) Collections

func (sink *Sink) Collections() []string

Collections returns the resource collections that this sink requests.

func (*Sink) ID

func (sink *Sink) ID() string

ID is the node id for this sink.

func (*Sink) Metadata

func (sink *Sink) Metadata() map[string]string

Metadata that is originally supplied when creating this sink.

func (*Sink) SnapshotRequestInfo

func (sink *Sink) SnapshotRequestInfo() []RecentRequestInfo

SnapshotRequestInfo returns a snapshot of the last known set of request results.

type Stream

type Stream interface {
	Send(*mcp.RequestResources) error
	Recv() (*mcp.Resources, error)
}

Stream is for sending RequestResources messages and receiving Resource messages.

type Updater

type Updater interface {
	// Apply is invoked when the node receives new configuration updates
	// from the server. The caller should return an error if any of the provided
	// configuration resources are invalid or cannot be applied. The node will
	// propagate errors back to the server accordingly.
	Apply(*Change) error
}

Updater provides configuration changes in batches of the same protobuf message type.