connector

Documentation

Overview

Package connector contains a reusable abstraction for efficiently watching for changes in resources in a Kubernetes cluster.

Index

• Constants
• Variables
• func ServiceInstanceID(name, addr string, httpPort, grpcPort int) string
• func WatchMeshConfigUpdated(msgBroker *messaging.Broker, stop <-chan struct{})
• type Backgrounder
• type Controller
  • func (c *Controller) HasSynced() bool
  • func (c *Controller) LastSyncResourceVersion() string
  • func (c *Controller) Run(stopCh <-chan struct{})
• type Event
• type Gateway
  • func (gw *Gateway) Enable() bool
• type ProtocolPort
• type Resource
• type ResourceDeleteFunc
• type ResourceUpsertFunc

Constants

const (
	//ConsulDiscoveryService defines consul discovery service name
	ConsulDiscoveryService = "consul"

	//EurekaDiscoveryService defines eureka discovery service name
	EurekaDiscoveryService = "eureka"

	//NacosDiscoveryService defines nacos discovery service name
	NacosDiscoveryService = "nacos"

	//MachineDiscoveryService defines machine discovery service name
	MachineDiscoveryService = "machine"
)

const (
	// AnnotationMeshServiceSync defines mesh service sync annotation
	AnnotationMeshServiceSync = "flomesh.io/mesh-service-sync"

	// AnnotationMeshServiceInternalSync defines mesh service internal sync annotation
	AnnotationMeshServiceInternalSync = "flomesh.io/mesh-service-internal-sync"

	// AnnotationCloudServiceInheritedFrom defines cloud service inherited annotation
	AnnotationCloudServiceInheritedFrom = "flomesh.io/cloud-service-inherited-from"

	// AnnotationCloudServiceInheritedClusterID defines cloud service cluster id annotation
	AnnotationCloudServiceInheritedClusterID = "flomesh.io/cloud-service-inherited-cluster-id"

	// AnnotationMeshEndpointAddr defines mesh endpoint addr annotation
	AnnotationMeshEndpointAddr = "flomesh.io/cloud-endpoint-addr"
)

const (
	// AnnotationServiceSyncK8sToCloud is the key of the annotation that determines
	// whether to sync the k8s Service to Consul/Eureka.
	AnnotationServiceSyncK8sToCloud = "flomesh.io/service-sync-k8s-to-cloud"

	// AnnotationServiceSyncK8sToFgw is the key of the annotation that determines
	// whether to sync the k8s Service to fsm gateway.
	AnnotationServiceSyncK8sToFgw = "flomesh.io/service-sync-k8s-to-fgw"

	// AnnotationServiceName is set to override the name of the service
	// registered. By default this will be the name of the CatalogService resource.
	AnnotationServiceName = "flomesh.io/service-name"

	// AnnotationServicePort specifies the port to use as the service instance
	// port when registering a service. This can be a named port in the
	// service or an integer value.
	AnnotationServicePort = "flomesh.io/service-port"

	// AnnotationServiceTags specifies the tags for the registered service
	// instance. Multiple tags should be comma separated. Whitespace around
	// the tags is automatically trimmed.
	AnnotationServiceTags = "flomesh.io/service-tags"

	// AnnotationServiceMetaPrefix is the prefix for setting meta key/value
	// for a service. The remainder of the key is the meta key.
	AnnotationServiceMetaPrefix = "flomesh.io/service-meta-"

	// AnnotationServiceWeight is the key of the annotation that determines
	// the traffic weight of the service which is spanned over multiple k8s cluster.
	// e.g. CatalogService `backend` in k8s cluster `A` receives 25% of the traffic
	// compared to same `backend` service in k8s cluster `B`.
	AnnotationServiceWeight = "flomesh.io/service-weight"
)

const (
	// ServiceSourceKey is the key used in the meta to track the "k8s" source.
	ServiceSourceKey = "fsm-connector-external-source"
)

Variables

var (
	// ServiceSourceValue is the value of the source.
	ServiceSourceValue = "sync-from-k8s"

	// ViaGateway defines gateway settings
	ViaGateway = &Gateway{}
)

var (
	GatewayAPIEnabled = false
)

var (
	ServiceInstanceIDFunc func(name, addr string, httpPort, grpcPort int) string
)

Functions

func ServiceInstanceID

func ServiceInstanceID(name, addr string, httpPort, grpcPort int) string

ServiceInstanceID generates a unique ID for a service. This ID is not meant to be particularly human-friendly.

func WatchMeshConfigUpdated

func WatchMeshConfigUpdated(msgBroker *messaging.Broker, stop <-chan struct{})

WatchMeshConfigUpdated watches update of meshconfig

Types

type Backgrounder

type Backgrounder interface {
	Run(<-chan struct{})
}

Backgrounder should be implemented by a Resource that requires additional background processing. If a Resource implements this, then the Controller will automatically Run the Backgrounder for the duration of the controller.

The channel will be closed when the Controller is quitting. The Controller will block until the Backgrounder completes.

type Controller

type Controller struct {
	Resource Resource
	// contains filtered or unexported fields
}

Controller is a generic cache.Controller implementation that watches Kubernetes for changes to specific set of resources and calls the configured callbacks as data changes.

func (*Controller) HasSynced

func (c *Controller) HasSynced() bool

HasSynced implements cache.Controller.

func (*Controller) LastSyncResourceVersion

func (c *Controller) LastSyncResourceVersion() string

LastSyncResourceVersion implements cache.Controller.

func (*Controller) Run

func (c *Controller) Run(stopCh <-chan struct{})

Run starts the Controller and blocks until stopCh is closed.

Important: Callers must ensure that Run is only called once at a time.

type Event

type Event struct {
	// Key is in the form of <namespace>/<name>, e.g. default/pod-abc123,
	// and corresponds to the resource modified.
	Key string
	// Obj holds the resource that was modified at the time of the event
	// occurring. If possible, the resource should be retrieved from the informer
	// cache, instead of using this field because the cache will be more up to
	// date at the time the event is processed.
	// In some cases, such as a delete event, the resource will no longer exist
	// in the cache and then it is useful to have the resource here.
	Obj interface{}
}

Event is something that occurred to the resources we're watching.

type Gateway

type Gateway struct {
	IngressIPSelector string
	EgressIPSelector  string

	IngressAddr string
	EgressAddr  string

	Ingress ProtocolPort
	Egress  ProtocolPort

	ClusterIP  string
	ExternalIP string
}

func (*Gateway) Enable

func (gw *Gateway) Enable() bool

type ProtocolPort

type ProtocolPort struct {
	HTTPPort uint
	GRPCPort uint
}

type Resource

type Resource interface {
	// Informer returns the SharedIndexInformer that the controller will
	// use to watch for changes. An Informer is the long-running task that
	// holds blocking queries to K8S and stores data in a local store.
	Informer() cache.SharedIndexInformer

	// Upsert is the callback called when processing the queue
	// of changes from the Informer. If an error is returned, the given item
	// will be retried.
	Upsert(key string, obj interface{}) error
	// Delete is called on object deletion.
	// obj is the last known state of the object before deletion. In some
	// cases, it may not be up to date with the latest state of the object.
	// If an error is returned, the given item will be retried.
	Delete(key string, obj interface{}) error
}

Resource should be implemented by anything that should be watchable by Controller. The Resource needs to be aware of how to create the Informer that is responsible for making API calls as well as what to do on Upsert and Delete.

type ResourceDeleteFunc

type ResourceDeleteFunc func(string, interface{}) error

type ResourceUpsertFunc

type ResourceUpsertFunc func(string, interface{}) error

ResourceUpsertFunc and ResourceDeleteFunc are the callback types for when a resource is inserted, updated, or deleted.