watcher

Documentation

Index

• type Metrics
  • func (m *Metrics) ClusterConfigInfo(sha string, info string)
  • func (m *Metrics) WatchBackoffDuration(d time.Duration)
  • func (m *Metrics) WatchClusterConfig(event string)
  • func (m *Metrics) WatchData(endpoint string)
  • func (m *Metrics) WatchErr(endpoint string, err error)
  • func (m *Metrics) WatchInit(d time.Duration)
• type Watcher
  • func NewWatcher(ctx context.Context, ...) (*Watcher, error)
  • func (w *Watcher) ConfigIPCount() int
  • func (w *Watcher) ConfigIPCount6() int
  • func (w *Watcher) EndpointCount() int
  • func (w *Watcher) GetEndpointAddressesForNode(nodeName string) []v1.EndpointAddress
  • func (w *Watcher) GetEndpointAddressesForService(serviceName string, namespace string, portName string) []v1.EndpointAddress
  • func (w *Watcher) GetLocalServiceWeight(nodeName string, namespace string, service string, portName string) float64
  • func (w *Watcher) GetPodIPsOnNode(nodeName string, serviceName string, namespace string, portName string) []string
  • func (w *Watcher) GetPortNumberForService(namespace string, serviceName string, portName string) int32
  • func (w *Watcher) HasConfigChanged(currentConfig *types.ClusterConfig, newConfig *types.ClusterConfig) bool
  • func (w *Watcher) NodeHasServiceRunning(nodeName string, namespace string, service string, portName string) bool
  • func (w *Watcher) ServiceCount() int
  • func (w *Watcher) ServiceDefinitionCount() int
  • func (w *Watcher) ServiceExistsInConfig(config *types.ClusterConfig, serviceName string, namespace string, ...) bool
  • func (w *Watcher) ServiceHasValidEndpoints(ns, svc string) bool
  • func (w *Watcher) ServiceIsConfigured(serviceName string, serviceNamespace string) bool
  • func (w *Watcher) Services() map[string]*v1.Service
  • func (w *Watcher) StartDebugWebServer()
  • func (w *Watcher) SubsetIPsForService(serviceName string, namespace string) []string
  • func (w *Watcher) VIPPoolCount() int
• type WatcherMetrics
  • func NewWatcherMetrics(kind, secZone string) WatcherMetrics

Types

type Metrics

type Metrics struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func (*Metrics) ClusterConfigInfo

func (m *Metrics) ClusterConfigInfo(sha string, info string)

func (*Metrics) WatchBackoffDuration

func (m *Metrics) WatchBackoffDuration(d time.Duration)

func (*Metrics) WatchClusterConfig

func (m *Metrics) WatchClusterConfig(event string)

func (*Metrics) WatchData

func (m *Metrics) WatchData(endpoint string)

func (*Metrics) WatchErr

func (m *Metrics) WatchErr(endpoint string, err error)

func (*Metrics) WatchInit

func (m *Metrics) WatchInit(d time.Duration)

type Watcher

type Watcher struct {
	sync.RWMutex

	ConfigMapNamespace string
	ConfigMapName      string
	ConfigKey          string

	AllServices   map[string]*v1.Service
	AllEndpoints  map[string]*v1.Endpoints
	AllPods       map[string]*v1.Pod
	AllPodsByNode map[string][]*v1.Pod // map of node name to pods on the node
	ConfigMap     *v1.ConfigMap

	// this is the 'official' configuration
	ClusterConfig *types.ClusterConfig
	Nodes         []*v1.Node

	// default listen services for vips in the vip pool
	AutoSvc  string
	AutoPort int
	// contains filtered or unexported fields
}

Watcher defines an interface for a ConfigMap containing the desired configuration state for the load balancer backend server. To generate the configmap, a watcher will collect both ConfigMap data from the kubernetes cluster as well as Endpoint data and it will joing these data sources together to create a derivative ConfigMap containing only services that are running on this specific node.

So, the way the watcher works is that it observes *both* the ConfigMap and the Endpoints sets for changes, and if any change is made to either, it generates a new ClusterConfig object internally. If the clusterconfig has changed from the prior configuration, we push it down the channel.

func NewWatcher

func NewWatcher(ctx context.Context, kubeConfigFile, cmNamespace, cmName, configKey, lbKind string, autoSvc string, autoPort int, logger log.FieldLogger) (*Watcher, error)

NewWatcher creates a new Watcher struct, which is used to watch services, endpoints, and more

func (*Watcher) ConfigIPCount

func (w *Watcher) ConfigIPCount() int

ConfigIPCount returns the number of v4 IPs in the cluster config

func (*Watcher) ConfigIPCount6

func (w *Watcher) ConfigIPCount6() int

ConfigIPCount6 returns the number of v6 IPs in the cluster config

func (*Watcher) EndpointCount

func (w *Watcher) EndpointCount() int

EndpointCount returns the number of endpoints known to the watcher

func (*Watcher) GetEndpointAddressesForNode

func (w *Watcher) GetEndpointAddressesForNode(nodeName string) []v1.EndpointAddress

GetEndpointAddressesForNode fetches all the subset addresses known by the watcher for a specific node.

func (*Watcher) GetEndpointAddressesForService

func (w *Watcher) GetEndpointAddressesForService(serviceName string, namespace string, portName string) []v1.EndpointAddress

GetEndpointAddressesForNodeAndPort fetches all the subset addresses known by the watcher for a specific node and service port name combination.

func (*Watcher) GetLocalServiceWeight

func (w *Watcher) GetLocalServiceWeight(nodeName string, namespace string, service string, portName string) float64

GetNodeServiceWeight computes the likelihood that any traffic for the service ends up on this particular node.

func (*Watcher) GetPodIPsOnNode

func (w *Watcher) GetPodIPsOnNode(nodeName string, serviceName string, namespace string, portName string) []string

GetPodIPsOnNode fetches all the PodIPs for the specified service on the specified node.

func (*Watcher) GetPortNumberForService

func (w *Watcher) GetPortNumberForService(namespace string, serviceName string, portName string) int32

GetPortNumberForService returns the port number for the service behind a VIP.

func (*Watcher) HasConfigChanged

func (w *Watcher) HasConfigChanged(currentConfig *types.ClusterConfig, newConfig *types.ClusterConfig) bool

HasConfigChanged determines if the cluster configuration has actually changed

func (*Watcher) NodeHasServiceRunning

func (w *Watcher) NodeHasServiceRunning(nodeName string, namespace string, service string, portName string) bool

NodeHasServiceRunning checks if the node has any endpoints (pods) running for a given service

func (*Watcher) ServiceCount

func (w *Watcher) ServiceCount() int

ServiceCount returns the number of services known to the watcher

func (*Watcher) ServiceDefinitionCount

func (w *Watcher) ServiceDefinitionCount() int

ServiceDefinitionCount returns the total number of PortConfig structs that exist currently in the watcher's known configuration

func (*Watcher) ServiceExistsInConfig

func (w *Watcher) ServiceExistsInConfig(config *types.ClusterConfig, serviceName string, namespace string, portName string) bool

func (*Watcher) ServiceHasValidEndpoints

func (w *Watcher) ServiceHasValidEndpoints(ns, svc string) bool

serviceHasValidEndpoints filters out any service that does not have an endpoint in its endpoints list. Kubernetes will remove these services from the kube-proxy, and we should, too.

func (*Watcher) ServiceIsConfigured

func (w *Watcher) ServiceIsConfigured(serviceName string, serviceNamespace string) bool

func (*Watcher) Services

func (w *Watcher) Services() map[string]*v1.Service

Services documented in interface definition

func (*Watcher) StartDebugWebServer

func (w *Watcher) StartDebugWebServer()

StartDebugWebService starts an http server for pprof and other debugging

func (*Watcher) SubsetIPsForService

func (w *Watcher) SubsetIPsForService(serviceName string, namespace string) []string

SubsetIPsForService returns the subset IPs that are currently cached in the watcher for the specified service

func (*Watcher) VIPPoolCount

func (w *Watcher) VIPPoolCount() int

VIPPoolCount returns the number of VIPs configured in total

type WatcherMetrics

type WatcherMetrics interface {
	// WatchBackoffDuration is a guage indicating the current length
	// of the backoff duration.
	WatchBackoffDuration(d time.Duration)

	// indicates that an error on initialization has occurred
	// counter rdel_lb_kube_connect_err_count
	WatchErr(endpoint string, err error)

	// indicates that the watcher has been reinitialized
	// counter rdel_lb_watch_init_count
	// bucket rdei_lb_watch_init_microseconds
	WatchInit(d time.Duration)

	// indicates how often new data arrives through each of the watch channels
	// counter rdei_lb_watch_data_count
	WatchData(endpoint string)

	// indicator of how often the cluster config is rebuilt and re-sent to the client
	// counter rdei_lb_watch_cluster_config_count
	WatchClusterConfig(event string)

	// contains the full applied configutration and a hash of it
	ClusterConfigInfo(sha string, info string)
}

func NewWatcherMetrics

func NewWatcherMetrics(kind, secZone string) WatcherMetrics

NewWatcherMetrics creates a new watcherMetrics struct