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
• type Aggregator
• type CreateSyncJob
  • func (job *CreateSyncJob) Run()
• type CtoKSource
  • func NewCtoKSource(controller connector.ConnectController, syncer *CtoKSyncer, ...) *CtoKSource
  • func (s *CtoKSource) Aggregate(ctx context.Context, svcName connector.MicroSvcName) map[connector.MicroSvcName]*connector.MicroSvcMeta
  • func (s *CtoKSource) Run(ctx context.Context)
• type CtoKSyncer
  • func NewCtoKSyncer(controller connector.ConnectController, ...) *CtoKSyncer
  • func (s *CtoKSyncer) Delete(key string, _ interface{}) error
  • func (s *CtoKSyncer) Informer() cache.SharedIndexInformer
  • func (s *CtoKSyncer) Ready()
  • func (s *CtoKSyncer) Run(ch <-chan struct{})
  • func (s *CtoKSyncer) SetMicroAggregator(microAggregator Aggregator)
  • func (s *CtoKSyncer) SetServices(svcs map[connector.MicroSvcName]connector.MicroSvcDomainName)
  • func (s *CtoKSyncer) Upsert(key string, raw interface{}) error
• type DeleteSyncJob
  • func (job *DeleteSyncJob) Run()
• type SyncJob
  • func (job *SyncJob) GetDoneCh() <-chan struct{}
  • func (job *SyncJob) JobName() string

Constants

const (
	// K8SQuietPeriod is the time to wait for no service changes before syncing.
	K8SQuietPeriod = messaging.ConnectorUpdateMaxWindow

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

	True  = "true"
	False = "false"
)

const (
	// CloudServiceLabel defines cloud service label
	CloudServiceLabel = "fsm-connector-cloud-service"
)

Functions

func IsSyncCloudNamespace

func IsSyncCloudNamespace(ns *corev1.Namespace) bool

IsSyncCloudNamespace if sync namespace

Types

type Aggregator

type Aggregator interface {
	// Aggregate micro services
	Aggregate(context.Context, connector.MicroSvcName) map[connector.MicroSvcName]*connector.MicroSvcMeta
}

Aggregator aggregates micro services

type CreateSyncJob

type CreateSyncJob struct {
	*SyncJob
	// contains filtered or unexported fields
}

CreateSyncJob is the job to sync

func (*CreateSyncJob) Run

func (job *CreateSyncJob) Run()

Run is the logic unit of job

type CtoKSource

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

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

func NewCtoKSource

func NewCtoKSource(controller connector.ConnectController,
	syncer *CtoKSyncer,
	discClient connector.ServiceDiscoveryClient,
	domain string) *CtoKSource

func (*CtoKSource) Aggregate

func (s *CtoKSource) Aggregate(ctx context.Context, svcName connector.MicroSvcName) map[connector.MicroSvcName]*connector.MicroSvcMeta

Aggregate micro services

func (*CtoKSource) Run

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

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

type CtoKSyncer

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

CtoKSyncer is the destination where services are registered.

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

func NewCtoKSyncer

func NewCtoKSyncer(
	controller connector.ConnectController,
	discClient connector.ServiceDiscoveryClient,
	kubeClient kubernetes.Interface,
	ctx context.Context,
	fsmNamespace string,
	nWorkers uint) *CtoKSyncer

NewCtoKSyncer creates a new mesh syncer

func (*CtoKSyncer) Delete

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

Delete implements the controller.Resource interface.

func (*CtoKSyncer) Informer

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

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

func (*CtoKSyncer) Ready

func (s *CtoKSyncer) Ready()

Ready wait util ready

func (*CtoKSyncer) Run

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

Run implements the controller.Backgrounder interface.

func (*CtoKSyncer) SetMicroAggregator

func (s *CtoKSyncer) SetMicroAggregator(microAggregator Aggregator)

func (*CtoKSyncer) SetServices

func (s *CtoKSyncer) SetServices(svcs map[connector.MicroSvcName]connector.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 (*CtoKSyncer) Upsert

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

Upsert implements the controller.Resource interface.

type DeleteSyncJob

type DeleteSyncJob struct {
	*SyncJob
	// contains filtered or unexported fields
}

DeleteSyncJob is the job to sync

func (*DeleteSyncJob) Run

func (job *DeleteSyncJob) Run()

Run is the logic unit of job

type SyncJob

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

SyncJob is the job to sync

func (*SyncJob) GetDoneCh

func (job *SyncJob) GetDoneCh() <-chan struct{}

GetDoneCh returns the channel, which when closed, indicates the job has been finished.

func (*SyncJob) JobName

func (job *SyncJob) JobName() string

JobName implementation for this job, for logging purposes