import "istio.io/istio/pkg/mcp/sink"
client_sink.go journal.go server_sink.go sink.go
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 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 (namespace/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
}Change is a collection of configuration objects of the same protobuf type.
Client implements the client for the MCP source service. The client is the sink and receives configuration from the server.
func NewClient(client mcp.ResourceSourceClient, options *Options) *Client
NewClient returns a new instance of Client.
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(names []string) []CollectionOptions
CollectionOptionsFromSlice returns a slice of collection options from a slice of collection names.
type InMemoryUpdater struct {
// contains filtered or unexported fields
}InMemoryUpdater is an implementation of Updater that keeps a simple in-memory state.
func NewInMemoryUpdater() *InMemoryUpdater
NewInMemoryUpdater returns a new instance of InMemoryUpdater
func (u *InMemoryUpdater) Apply(c *Change) error
Apply the change to the InMemoryUpdater.
func (u *InMemoryUpdater) Get(collection string) []*Object
Get current state for the given collection.
Object contains a decoded versioned object with metadata received from the server.
type Options struct {
CollectionOptions []CollectionOptions
Updater Updater
ID string
Metadata map[string]string
Reporter monitoring.Reporter
}Options contains options for configuring MCP sinks.
RateLimiter is partially representing standard lib's rate limiter
type RecentRequestInfo struct {
Time time.Time
Request *mcp.RequestResources
}RecentRequestInfo is metadata about a request that the client has sent.
func (r RecentRequestInfo) Acked() bool
Acked indicates whether the message was an ack or not.
type RecentRequestsJournal struct {
// contains filtered or unexported fields
}RecentRequestsJournal captures debug metadata about the latest requests that was sent by this client.
func NewRequestJournal() *RecentRequestsJournal
func (r *RecentRequestsJournal) RecordRequestResources(req *mcp.RequestResources)
func (r *RecentRequestsJournal) Snapshot() []RecentRequestInfo
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(sinkOptions *Options, serverOptions *ServerOptions) *Server
NewServer creates a new instance of a MCP sink server.
func (s *Server) EstablishResourceStream(stream mcp.ResourceSink_EstablishResourceStreamServer) error
EstablishResourceStream implements the ResourceSinkServer interface.
type ServerOptions struct {
AuthChecker AuthChecker
RateLimiter rate.Limit
}ServerOptions contains source server specific options
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.
New creates a new resource sink.
Collections returns the resource collections that this sink requests.
ID is the node id for this sink.
Metadata that is originally supplied when creating this sink.
ProcessStream implements the MCP message exchange for the resource sink. It accepts the sink stream interface and returns when a send or receive error occurs. The caller is responsible for handling gRPC client/server specific error handling.
func (sink *Sink) SnapshotRequestInfo() []RecentRequestInfo
SnapshotRequestInfo returns a snapshot of the last known set of request results.
Stream is for sending RequestResources messages and receiving Resource messages.
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.
Package sink imports 17 packages (graph) and is imported by 19 packages. Updated 2019-10-22. Refresh now. Tools for package owners.