worker

README

Temporal Worker

Temporal Worker is a role for Temporal service used for hosting any components responsible for performing background processing on the Temporal cluster.

Replicator

Replicator is a background worker responsible for consuming replication tasks generated by remote Temporal clusters and pass it down to processor, so they can be applied to local Temporal cluster.

Quickstart for localhost development

1. Start Temporal development server for active zone:

   make start-cdc-active
   

2. Start Temporal development server for standby(passive) zone:

   make start-cdc-standby
   

3. Connect two Temporal clusters:

   tctl --ad 127.0.0.1:7233 adm cl upsert-remote-cluster --frontend_address "localhost:8233"
   tctl --ad 127.0.0.1:8233 adm cl upsert-remote-cluster --frontend_address "localhost:7233"
   

4. Create global namespaces

   tctl --ns sample namespace register --gd true --ac active --cl active standby
   

5. Failover between zones:

   Failover to standby:

   tctl --ns sample namespace update --ac standby
   

   Failback to active:

   tctl --ns sample namespace update --ac active
   

Archiver

Archiver is used to handle archival of workflow execution histories. It does this by hosting a Temporal client worker and running an archival system workflow. The archival client gets used to initiate archival through signal sending. The archiver shards work across several workflows.

Documentation

Index

• Constants
• Variables
• func FrontendClientProvider(cfg *config.Config, clientFactory client.Factory) (workflowservice.WorkflowServiceClient, error)
• func NewPerNamespaceWorkerManager(params perNamespaceWorkerManagerInitParams) *perNamespaceWorkerManager
• func NewWorkerManager(params initParams) *workerManager
• func PersistenceRateLimitingParamsProvider(serviceConfig *Config) service.PersistenceRateLimitingParams
• func ServiceLifetimeHooks(lc fx.Lifecycle, svcStoppedCh chan struct{}, svc *Service)
• func ThrottledLoggerRpsFnProvider(serviceConfig *Config) resource.ThrottledLoggerRpsFn
• func VisibilityManagerProvider(logger log.Logger, metricsClient metrics.Client, ...) (manager.VisibilityManager, error)
• type Config
  • func ConfigProvider(dc *dynamicconfig.Collection, persistenceConfig *config.Persistence) *Config
  • func NewConfig(dc *dynamicconfig.Collection, persistenceConfig *config.Persistence, ...) *Config
• type Service
  • func NewService(logger resource.SnTaggedLogger, serviceConfig *Config, ...) (*Service, error)
  • func (s *Service) Start()
  • func (s *Service) Stop()

Constants

const DefaultWorkerTaskQueue = "default-worker-tq"

Variables

var Module = fx.Options(
	migration.Module,
	addsearchattributes.Module,
	resource.Module,
	deletenamespace.Module,
	scheduler.Module,
	fx.Provide(
		fx.Annotated{
			Name:   localFrontendClient,
			Target: FrontendClientProvider,
		},
	),
	fx.Provide(VisibilityManagerProvider),
	fx.Provide(dynamicconfig.NewCollection),
	fx.Provide(ThrottledLoggerRpsFnProvider),
	fx.Provide(ConfigProvider),
	fx.Provide(PersistenceRateLimitingParamsProvider),
	fx.Provide(NewService),
	fx.Provide(NewWorkerManager),
	fx.Provide(NewPerNamespaceWorkerManager),
	fx.Invoke(ServiceLifetimeHooks),
)

Functions

func FrontendClientProvider

func FrontendClientProvider(
	cfg *config.Config,
	clientFactory client.Factory,
) (workflowservice.WorkflowServiceClient, error)

func NewPerNamespaceWorkerManager

func NewPerNamespaceWorkerManager(params perNamespaceWorkerManagerInitParams) *perNamespaceWorkerManager

func NewWorkerManager

func NewWorkerManager(params initParams) *workerManager

func PersistenceRateLimitingParamsProvider

func PersistenceRateLimitingParamsProvider(
	serviceConfig *Config,
) service.PersistenceRateLimitingParams

func ServiceLifetimeHooks

func ServiceLifetimeHooks(
	lc fx.Lifecycle,
	svcStoppedCh chan struct{},
	svc *Service,
)

func ThrottledLoggerRpsFnProvider

func ThrottledLoggerRpsFnProvider(serviceConfig *Config) resource.ThrottledLoggerRpsFn

func VisibilityManagerProvider

func VisibilityManagerProvider(
	logger log.Logger,
	metricsClient metrics.Client,
	persistenceConfig *config.Persistence,
	serviceConfig *Config,
	esConfig *esclient.Config,
	esClient esclient.Client,
	persistenceServiceResolver resolver.ServiceResolver,
	searchAttributesMapper searchattribute.Mapper,
	saProvider searchattribute.Provider,
) (manager.VisibilityManager, error)

Types

type Config

type Config struct {
	ArchiverConfig                        *archiver.Config
	ScannerCfg                            *scanner.Config
	ParentCloseCfg                        *parentclosepolicy.Config
	BatcherCfg                            *batcher.Config
	ThrottledLogRPS                       dynamicconfig.IntPropertyFn
	PersistenceMaxQPS                     dynamicconfig.IntPropertyFn
	PersistenceGlobalMaxQPS               dynamicconfig.IntPropertyFn
	EnablePersistencePriorityRateLimiting dynamicconfig.BoolPropertyFn
	EnableBatcher                         dynamicconfig.BoolPropertyFn
	EnableParentClosePolicyWorker         dynamicconfig.BoolPropertyFn
	PerNamespaceWorkerCount               dynamicconfig.IntPropertyFnWithNamespaceFilter

	StandardVisibilityPersistenceMaxReadQPS   dynamicconfig.IntPropertyFn
	StandardVisibilityPersistenceMaxWriteQPS  dynamicconfig.IntPropertyFn
	AdvancedVisibilityPersistenceMaxReadQPS   dynamicconfig.IntPropertyFn
	AdvancedVisibilityPersistenceMaxWriteQPS  dynamicconfig.IntPropertyFn
	EnableReadVisibilityFromES                dynamicconfig.BoolPropertyFnWithNamespaceFilter
	EnableReadFromSecondaryAdvancedVisibility dynamicconfig.BoolPropertyFnWithNamespaceFilter
	VisibilityDisableOrderByClause            dynamicconfig.BoolPropertyFn
}

Config contains all the service config for worker

func ConfigProvider

func ConfigProvider(
	dc *dynamicconfig.Collection,
	persistenceConfig *config.Persistence,
) *Config

func NewConfig

func NewConfig(dc *dynamicconfig.Collection, persistenceConfig *config.Persistence, enableReadFromES bool) *Config

NewConfig builds the new Config for worker service

type Service

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

Service represents the temporal-worker service. This service hosts all background processing needed for temporal cluster: Replicator: Handles applying replication tasks generated by remote clusters. Archiver: Handles archival of workflow histories.

func NewService

func NewService(
	logger resource.SnTaggedLogger,
	serviceConfig *Config,
	sdkClientFactory sdk.ClientFactory,
	esClient esclient.Client,
	archivalMetadata carchiver.ArchivalMetadata,
	clusterMetadata cluster.Metadata,
	metricsClient metrics.Client,
	clientBean client.Bean,
	clusterMetadataManager persistence.ClusterMetadataManager,
	namespaceRegistry namespace.Registry,
	executionManager persistence.ExecutionManager,
	archiverProvider provider.ArchiverProvider,
	persistenceBean persistenceClient.Bean,
	membershipMonitor membership.Monitor,
	namespaceReplicationQueue persistence.NamespaceReplicationQueue,
	metricsScope metrics.UserScope,
	metadataManager persistence.MetadataManager,
	taskManager persistence.TaskManager,
	historyClient historyservice.HistoryServiceClient,
	workerManager *workerManager,
	perNamespaceWorkerManager *perNamespaceWorkerManager,
	visibilityManager manager.VisibilityManager,
) (*Service, error)

func (*Service) Start

func (s *Service) Start()

Start is called to start the service

func (*Service) Stop

func (s *Service) Stop()

Stop is called to stop the service