ctok

Documentation

Overview

Package ctok implements a syncer from cloud to k8s.

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

Index

• Constants
• func IsSyncCloudNamespace(ns *corev1.Namespace) bool
• func SetSyncCloudNamespace(ns string)
• func WithGateway(enable bool)
• type Aggregator
• type MicroEndpointAddr
  • func (addr MicroEndpointAddr) To16() net.IP
  • func (addr MicroEndpointAddr) To4() net.IP
• type MicroSvcAppProtocol
• type MicroSvcDomainName
• type MicroSvcMeta
• type MicroSvcName
• type MicroSvcPort
• type Sink
  • func NewSink(ctx context.Context, kubeClient kubernetes.Interface, ...) *Sink
  • func (s *Sink) Delete(key string, _ interface{}) error
  • func (s *Sink) Informer() cache.SharedIndexInformer
  • func (s *Sink) Ready()
  • func (s *Sink) Run(ch <-chan struct{})
  • func (s *Sink) SetServices(svcs map[MicroSvcName]MicroSvcDomainName)
  • func (s *Sink) Upsert(key string, raw interface{}) error
• type Source
  • func (s *Source) Aggregate(svcName MicroSvcName, svcDomainName MicroSvcDomainName) map[MicroSvcName]*MicroSvcMeta
  • func (s *Source) Run(ctx context.Context)

Constants

const (
	// K8SQuietPeriod is the time to wait for no service changes before syncing.
	K8SQuietPeriod = 1 * time.Second

	// K8SMaxPeriod is the maximum time to wait before forcing a sync, even
	// if there are active changes going on.
	K8SMaxPeriod = 5 * time.Second

	True  = "true"
	False = "false"
)

const (
	// CloudSourcedServiceLabel defines cloud-sourced service label
	CloudSourcedServiceLabel = "cloud-sourced-service"
	// CloudServiceLabel defines cloud service label
	CloudServiceLabel = "cloud-service"
)

Functions

func IsSyncCloudNamespace

func IsSyncCloudNamespace(ns *corev1.Namespace) bool

IsSyncCloudNamespace if sync namespace

func SetSyncCloudNamespace

func SetSyncCloudNamespace(ns string)

SetSyncCloudNamespace sets sync namespace

func WithGateway

func WithGateway(enable bool)

WithGateway sets enable or disable

Types

type Aggregator

type Aggregator interface {
	// Aggregate micro services
	Aggregate(svcName MicroSvcName, svcDomainName MicroSvcDomainName) map[MicroSvcName]*MicroSvcMeta
}

Aggregator aggregates micro services

type MicroEndpointAddr

type MicroEndpointAddr string

MicroEndpointAddr defines string as micro endpoint addr

func (MicroEndpointAddr) To16

func (addr MicroEndpointAddr) To16() net.IP

To16 converts the IP address ip to a 16-byte representation. If ip is not an IP address (it is the wrong length), To16 returns nil.

func (MicroEndpointAddr) To4

func (addr MicroEndpointAddr) To4() net.IP

To4 converts the IPv4 address ip to a 4-byte representation. If ip is not an IPv4 address, To4 returns nil.

type MicroSvcAppProtocol

type MicroSvcAppProtocol string

MicroSvcAppProtocol defines app protocol

type MicroSvcDomainName

type MicroSvcDomainName string

MicroSvcDomainName defines string as microservice domain name

type MicroSvcMeta

type MicroSvcMeta struct {
	Ports       map[MicroSvcPort]MicroSvcAppProtocol
	Addresses   map[MicroEndpointAddr]int
	ClusterId   string
	HealthCheck bool
}

MicroSvcMeta defines micro service meta

type MicroSvcName

type MicroSvcName string

MicroSvcName defines string as microservice name

type MicroSvcPort

type MicroSvcPort int

MicroSvcPort defines int as micro service port

type Sink

type Sink struct {
	KubeClient kubernetes.Interface

	DiscClient provider.ServiceDiscoveryClient

	MicroAggregator Aggregator

	// SyncPeriod is the duration to wait between registering or deregistering
	// services in Kubernetes. This can be fairly short since no work will be
	// done if there are no changes.
	SyncPeriod time.Duration

	// Ctx is used to cancel the Sink.
	Ctx context.Context
	// contains filtered or unexported fields
}

Sink is the destination where services are registered.

While in practice we only have one sink (K8S), the interface abstraction makes it easy and possible to test the Source in isolation.

func NewSink

func NewSink(ctx context.Context, kubeClient kubernetes.Interface, discClient provider.ServiceDiscoveryClient, fsmNamespace string) *Sink

NewSink creates a new mesh sink

func (*Sink) Delete

func (s *Sink) Delete(key string, _ interface{}) error

Delete implements the controller.Resource interface.

func (*Sink) Informer

func (s *Sink) Informer() cache.SharedIndexInformer

Informer implements the controller.Resource interface. It tells Kubernetes that we want to watch for changes to Services.

func (*Sink) Ready

func (s *Sink) Ready()

Ready wait util ready

func (*Sink) Run

func (s *Sink) Run(ch <-chan struct{})

Run implements the controller.Backgrounder interface.

func (*Sink) SetServices

func (s *Sink) SetServices(svcs map[MicroSvcName]MicroSvcDomainName)

SetServices is called with the services that should be created. The key is the service name and the destination is the external DNS entry to point to.

func (*Sink) Upsert

func (s *Sink) Upsert(key string, raw interface{}) error

Upsert implements the controller.Resource interface.

type Source

type Source struct {
	DiscClient  provider.ServiceDiscoveryClient
	Domain      string // DNS domain
	Sink        *Sink  // Sink is the sink to update with services
	Prefix      string // Prefix is a prefix to prepend to services
	FilterTag   string // The tag value for services registered
	PrefixTag   string
	SuffixTag   string
	PassingOnly bool
}

Source is the source for the sync that watches cloud services and updates a Sink whenever the set of services to register changes.

func (*Source) Aggregate

func (s *Source) Aggregate(svcName MicroSvcName, svcDomainName MicroSvcDomainName) map[MicroSvcName]*MicroSvcMeta

Aggregate micro services

func (*Source) Run

func (s *Source) Run(ctx context.Context)

Run is the long-running loop for watching cloud services and updating the Sink.