temporal

package
v1.25.0-113.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 18, 2024 License: MIT Imports: 57 Imported by: 15

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// Services is the set of all valid temporal services as strings (needs to be strings to
	// keep ServerOptions interface stable)
	Services = []string{
		string(primitives.FrontendService),
		string(primitives.InternalFrontendService),
		string(primitives.HistoryService),
		string(primitives.MatchingService),
		string(primitives.WorkerService),
	}

	// DefaultServices is the set of services to start by default if services are not given on
	// the command line.
	DefaultServices = []string{
		string(primitives.FrontendService),
		string(primitives.HistoryService),
		string(primitives.MatchingService),
		string(primitives.WorkerService),
	}
)
View Source 
var FxLogAdapter = fx.WithLogger(func(logger log.Logger) fxevent.Logger {
	return &fxLogAdapter{logger: logger}
})
View Source 
var ServiceTracingModule = fx.Options(
	fx.Supply([]otelsdktrace.BatchSpanProcessorOption{}),
	fx.Provide(
		fx.Annotate(
			func(exps []otelsdktrace.SpanExporter, opts []otelsdktrace.BatchSpanProcessorOption) []otelsdktrace.SpanProcessor {
				sps := make([]otelsdktrace.SpanProcessor, 0, len(exps))
				for _, exp := range exps {
					sps = append(sps, otelsdktrace.NewBatchSpanProcessor(exp, opts...))
				}
				return sps
			},
			fx.ParamTags(`optional:"true"`, ``),
		),
	),
	fx.Provide(
		fx.Annotate(
			func(rsn primitives.ServiceName, rsi resource.InstanceID) (*otelresource.Resource, error) {
				attrs := []attribute.KeyValue{
					semconv.ServiceNameKey.String(telemetry.ResourceServiceName(rsn, os.LookupEnv)),
					semconv.ServiceVersionKey.String(headers.ServerVersion),
				}
				if rsi != "" {
					attrs = append(attrs, semconv.ServiceInstanceIDKey.String(string(rsi)))
				}

				return otelresource.New(context.Background(),
					otelresource.WithProcess(),
					otelresource.WithOS(),
					otelresource.WithHost(),
					otelresource.WithContainer(),
					otelresource.WithAttributes(attrs...),
				)
			},
			fx.ParamTags(``, `optional:"true"`),
		),
	),
	fx.Provide(
		func(r *otelresource.Resource, sps []otelsdktrace.SpanProcessor) []otelsdktrace.TracerProviderOption {
			opts := make([]otelsdktrace.TracerProviderOption, 0, len(sps)+1)
			opts = append(opts, otelsdktrace.WithResource(r))
			for _, sp := range sps {
				opts = append(opts, otelsdktrace.WithSpanProcessor(sp))
			}
			return opts
		},
	),
	fx.Provide(func(lc fx.Lifecycle, opts []otelsdktrace.TracerProviderOption) trace.TracerProvider {
		tp := otelsdktrace.NewTracerProvider(opts...)
		lc.Append(fx.Hook{OnStop: func(ctx context.Context) error {
			return tp.Shutdown(ctx)
		}})
		return tp
	}),

	fx.Provide(func() propagation.TextMapPropagator { return propagation.TraceContext{} }),
	fx.Provide(telemetry.NewServerTraceInterceptor),
	fx.Provide(telemetry.NewClientTraceInterceptor),
)

ServiceTracingModule holds per-service (i.e. frontend/history/matching/worker) fx state. The following types can be overriden with fx.Replace/fx.Decorate:

  • []go.opentelemetry.io/otel/sdk/trace.BatchSpanProcessorOption default: empty slice
  • []go.opentelemetry.io/otel/sdk/trace.SpanProcessor default: wrap each otelsdktrace.SpanExporter with otelsdktrace.NewBatchSpanProcessor
  • *go.opentelemetry.io/otel/sdk/resource.Resource default: resource.Default() augmented with the supplied serviceName
  • []go.opentelemetry.io/otel/sdk/trace.TracerProviderOption default: the provided resource.Resource and each of the otelsdktrace.SpanExporter
  • go.opentelemetry.io/otel/trace.TracerProvider default: otelsdktrace.NewTracerProvider with each of the otelsdktrace.TracerProviderOption
  • go.opentelemetry.io/otel/ppropagation.TextMapPropagator default: propagation.TraceContext{}
  • telemetry.ServerTraceInterceptor
  • telemetry.ClientTraceInterceptor
View Source 
var (
	TopLevelModule = fx.Options(
		fx.Provide(
			NewServerFxImpl,
			ServerOptionsProvider,
			dynamicconfig.NewCollection,
			resource.ArchivalMetadataProvider,
			TaskCategoryRegistryProvider,
			PersistenceFactoryProvider,
			HistoryServiceProvider,
			MatchingServiceProvider,
			FrontendServiceProvider,
			InternalFrontendServiceProvider,
			WorkerServiceProvider,
			ApplyClusterMetadataConfigProvider,
		),
		pprof.Module,
		TraceExportModule,
		FxLogAdapter,
		fx.Invoke(ServerLifetimeHooks),
	)
)
View Source 
var TraceExportModule = fx.Options(
	fx.Invoke(func(log log.Logger) {
		otel.SetErrorHandler(otel.ErrorHandlerFunc(
			func(err error) {
				log.Warn("OTEL error", tag.Error(err), tag.ServiceErrorType(err))
			}),
		)
	}),

	fx.Provide(func(lc fx.Lifecycle, c *config.Config) ([]otelsdktrace.SpanExporter, error) {
		exportersByType, err := c.ExporterConfig.SpanExporters()
		if err != nil {
			return nil, err
		}

		exportersByTypeFromEnv, err := telemetry.SpanExportersFromEnv(os.LookupEnv)
		if err != nil {
			return nil, err
		}

		maps.Copy(exportersByType, exportersByTypeFromEnv)

		exporters := maps.Values(exportersByType)
		lc.Append(fx.Hook{
			OnStart: startAll(exporters),
			OnStop:  shutdownAll(exporters),
		})
		return exporters, nil
	}),
)

TraceExportModule holds process-global telemetry fx state defining the set of OTEL trace/span exporters used by tracing instrumentation. The following types can be overriden/augmented with fx.Replace/fx.Decorate:

- []go.opentelemetry.io/otel/sdk/trace.SpanExporter

Functions

func ApplyClusterMetadataConfigProvider added in v1.14.0 

func ApplyClusterMetadataConfigProvider(
	logger log.Logger,
	svc *config.Config,
	persistenceServiceResolver resolver.ServiceResolver,
	persistenceFactoryProvider persistenceClient.FactoryProviderFn,
	customDataStoreFactory persistenceClient.AbstractDataStoreFactory,
	metricsHandler metrics.Handler,
) (*cluster.Config, config.Persistence, error)

ApplyClusterMetadataConfigProvider performs a config check against the configured persistence store for cluster metadata. If there is a mismatch, the persisted values take precedence and will be written over in the config objects. This is to keep this check hidden from downstream calls. TODO: move this to cluster.fx

func InterruptCh 

func InterruptCh() <-chan interface{}

func PersistenceFactoryProvider added in v1.16.0 

func PersistenceFactoryProvider() persistenceClient.FactoryProviderFn

func ServerLifetimeHooks added in v1.14.0 

func ServerLifetimeHooks(
	lc fx.Lifecycle,
	svr *ServerImpl,
)

func ServerOptionsProvider added in v1.14.0 

func ServerOptionsProvider(opts []ServerOption) (serverOptionsProvider, error)

func TaskCategoryRegistryProvider added in v1.23.0 

func TaskCategoryRegistryProvider(archivalMetadata archiver.ArchivalMetadata, dc *dynamicconfig.Collection) tasks.TaskCategoryRegistry

TaskCategoryRegistryProvider provides an immutable tasks.TaskCategoryRegistry to the server, which is intended to be shared by each service. Why do we need to initialize this at the top-level? Because, even though the presence of the archival task category is only needed by the history service, which must conditionally start a queue processor for it, we also do validation on request task categories in the frontend service. As a result, we need to initialize the registry in the server graph, and then propagate it to the service graphs. Otherwise, it would be isolated to the history service's graph.

Types

type ClusterMetadataLoader added in v1.24.0 

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

ClusterMetadataLoader loads cluster metadata from the database and merges it with the static config. TODO: move this to the cluster package. It is here temporarily to avoid a circular dependency.

func NewClusterMetadataLoader added in v1.24.0 

func NewClusterMetadataLoader(manager persistence.ClusterMetadataManager, logger log.Logger) *ClusterMetadataLoader

NewClusterMetadataLoader creates a new ClusterMetadataLoader that loads cluster metadata from the database.

func (*ClusterMetadataLoader) LoadAndMergeWithStaticConfig added in v1.24.0 

func (c *ClusterMetadataLoader) LoadAndMergeWithStaticConfig(ctx context.Context, svc *config.Config) error

LoadAndMergeWithStaticConfig loads cluster metadata from the database and merges it with the static config.

type Server 

type Server interface {
	Start() error
	Stop() error
}

func NewServer 

func NewServer(opts ...ServerOption) (Server, error)

NewServer returns a new instance of server that serves one or many services.

type ServerFx added in v1.14.0 

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

func NewServerFx added in v1.14.0 

func NewServerFx(topLevelModule fx.Option, opts ...ServerOption) (*ServerFx, error)

func (*ServerFx) Start added in v1.14.0 

func (s *ServerFx) Start() error

Start temporal server. This function should be called only once, Server doesn't support multiple restarts.

func (*ServerFx) Stop added in v1.14.0 

func (s *ServerFx) Stop() error

Stop stops the server.

type ServerImpl added in v1.14.0 

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

ServerImpl is temporal server.

func NewServerFxImpl added in v1.14.0 

func NewServerFxImpl(
	opts *serverOptions,
	logger log.Logger,
	namespaceLogger resource.NamespaceLogger,
	stoppedCh chan interface{},
	servicesGroup ServicesGroupIn,
	persistenceConfig config.Persistence,
	clusterMetadata *cluster.Config,
	persistenceFactoryProvider persistenceClient.FactoryProviderFn,
	metricsHandler metrics.Handler,
) *ServerImpl

NewServerFxImpl returns a new instance of server that serves one or many services.

func (*ServerImpl) Start added in v1.14.0 

func (s *ServerImpl) Start(ctx context.Context) error

func (*ServerImpl) Stop added in v1.14.0 

func (s *ServerImpl) Stop(ctx context.Context) error

type ServerOption 

type ServerOption interface {
	// contains filtered or unexported methods
}

func ForServices 

func ForServices(names []string) ServerOption

ForServices indicates which supplied services (e.g. frontend, history, matching, worker) within the server to start

func InterruptOn 

func InterruptOn(interruptCh <-chan interface{}) ServerOption

InterruptOn interrupts server on the signal from server. If channel is nil Start() will block forever.

func WithAudienceGetter added in v1.10.3 

func WithAudienceGetter(audienceGetter func(cfg *config.Config) authorization.JWTAudienceMapper) ServerOption

WithAudienceGetter configures JWT audience getter for authorization

func WithAuthorizer 

func WithAuthorizer(authorizer authorization.Authorizer) ServerOption

WithAuthorizer sets a low level authorizer to allow/deny all API calls

func WithChainedFrontendGrpcInterceptors added in v1.14.0 

func WithChainedFrontendGrpcInterceptors(
	interceptors ...grpc.UnaryServerInterceptor,
) ServerOption

WithChainedFrontendGrpcInterceptors sets a chain of ordered custom grpc interceptors that will be invoked for all Frontend gRPC API calls. The list of custom interceptors will be appended to the end of the internal ServerInterceptors. The custom interceptors will be invoked in the order as they appear in the supplied list, after the internal ServerInterceptors.

func WithClaimMapper added in v1.4.0 

func WithClaimMapper(claimMapper func(cfg *config.Config) authorization.ClaimMapper) ServerOption

WithClaimMapper configures a role mapper for authorization

func WithClientFactoryProvider added in v1.11.0 

func WithClientFactoryProvider(clientFactoryProvider client.FactoryProvider) ServerOption

WithClientFactoryProvider sets a custom ClientFactoryProvider NOTE: this option is experimental and may be changed or removed in future release.

func WithConfig 

func WithConfig(cfg *config.Config) ServerOption

WithConfig sets a custom configuration

func WithConfigLoader 

func WithConfigLoader(configDir string, env string, zone string) ServerOption

WithConfigLoader sets a custom configuration load

func WithCustomDataStoreFactory added in v1.11.0 

func WithCustomDataStoreFactory(customFactory persistenceclient.AbstractDataStoreFactory) ServerOption

WithCustomDataStoreFactory sets a custom AbstractDataStoreFactory NOTE: this option is experimental and may be changed or removed in future release.

func WithCustomMetricsHandler added in v1.17.0 

func WithCustomMetricsHandler(provider metrics.Handler) ServerOption

WithCustomerMetricsProvider sets a custom implementation of the metrics.MetricsHandler interface metrics.MetricsHandler is the base interface for publishing metric events

func WithCustomVisibilityStoreFactory added in v1.23.0 

func WithCustomVisibilityStoreFactory(customFactory visibility.VisibilityStoreFactory) ServerOption

func WithDynamicConfigClient added in v1.5.7 

func WithDynamicConfigClient(c dynamicconfig.Client) ServerOption

WithDynamicConfigClient sets custom client for reading dynamic configuration.

func WithElasticsearchHttpClient added in v1.5.7 

func WithElasticsearchHttpClient(c *http.Client) ServerOption

WithElasticsearchHttpClient sets a custom HTTP client which is used to make requests to Elasticsearch

func WithLogger added in v1.5.7 

func WithLogger(logger log.Logger) ServerOption

WithLogger sets a custom logger

func WithNamespaceLogger added in v1.11.0 

func WithNamespaceLogger(namespaceLogger log.Logger) ServerOption

WithNamespaceLogger sets an optional logger for all frontend operations

func WithPersistenceServiceResolver added in v1.5.7 

func WithPersistenceServiceResolver(r resolver.ServiceResolver) ServerOption

WithPersistenceServiceResolver sets a custom persistence service resolver which will convert service name or address value from config to another address

func WithSearchAttributesMapper added in v1.12.1 

func WithSearchAttributesMapper(m searchattribute.Mapper) ServerOption

WithSearchAttributesMapper sets a custom search attributes mapper which converts search attributes aliases to field names and vice versa.

func WithStaticHosts added in v1.24.0 

func WithStaticHosts(hostsByService map[primitives.ServiceName]static.Hosts) ServerOption

WithStaticHosts disables dynamic service membership and resolves service hosts statically. At least one host must be provided for all required services (frontend, history, matching, worker). And a self-address must be provided for all services passed to ForServices.

func WithTLSConfigFactory 

func WithTLSConfigFactory(tlsConfigProvider encryption.TLSConfigProvider) ServerOption

WithTLSConfigFactory overrides default provider of TLS configuration

type ServiceProviderParamsCommon added in v1.14.0 

type ServiceProviderParamsCommon struct {
	fx.In

	Cfg                        *config.Config
	ServiceNames               resource.ServiceNames
	Logger                     log.Logger
	NamespaceLogger            resource.NamespaceLogger
	DynamicConfigClient        dynamicconfig.Client
	MetricsHandler             metrics.Handler
	EsConfig                   *esclient.Config
	EsClient                   esclient.Client
	TlsConfigProvider          encryption.TLSConfigProvider
	PersistenceConfig          config.Persistence
	ClusterMetadata            *cluster.Config
	ClientFactoryProvider      client.FactoryProvider
	AudienceGetter             authorization.JWTAudienceMapper
	PersistenceServiceResolver resolver.ServiceResolver
	PersistenceFactoryProvider persistenceClient.FactoryProviderFn
	SearchAttributesMapper     searchattribute.Mapper
	CustomFrontendInterceptors []grpc.UnaryServerInterceptor
	Authorizer                 authorization.Authorizer
	ClaimMapper                authorization.ClaimMapper
	DataStoreFactory           persistenceClient.AbstractDataStoreFactory
	VisibilityStoreFactory     visibility.VisibilityStoreFactory
	SpanExporters              []otelsdktrace.SpanExporter
	InstanceID                 resource.InstanceID                     `optional:"true"`
	StaticServiceHosts         map[primitives.ServiceName]static.Hosts `optional:"true"`
	TaskCategoryRegistry       tasks.TaskCategoryRegistry
}

func (ServiceProviderParamsCommon) GetCommonServiceOptions added in v1.23.0 

func (params ServiceProviderParamsCommon) GetCommonServiceOptions(serviceName primitives.ServiceName) fx.Option

GetCommonServiceOptions returns an fx.Option which combines all the common dependencies for a service. Why do we need this? We are propagating dependencies from one fx graph to another. This is not ideal, since we should instead either have one shared fx graph, or propagate the individual fx options. So, in the server graph, dependencies exist as fx providers, and, in the process of building the server graph, we build the individual service graphs. We realize the dependencies in the server graph implicitly into the ServiceProviderParamsCommon object. Then, we convert them back into fx providers here. Essentially, we want an `fx.In` object in the server graph, and an `fx.Out` object in the service graphs. This is a workaround to achieve something similar.

type ServicesGroupIn added in v1.14.0 

type ServicesGroupIn struct {
	fx.In
	Services []*ServicesMetadata `group:"services"`
}

type ServicesGroupOut added in v1.14.0 

type ServicesGroupOut struct {
	fx.Out
	Services *ServicesMetadata `group:"services"`
}

func FrontendServiceProvider added in v1.14.0 

func FrontendServiceProvider(
	params ServiceProviderParamsCommon,
) (ServicesGroupOut, error)

func HistoryServiceProvider added in v1.14.0 

func HistoryServiceProvider(
	params ServiceProviderParamsCommon,
) (ServicesGroupOut, error)

func InternalFrontendServiceProvider added in v1.20.0 

func InternalFrontendServiceProvider(
	params ServiceProviderParamsCommon,
) (ServicesGroupOut, error)

func MatchingServiceProvider added in v1.14.0 

func MatchingServiceProvider(
	params ServiceProviderParamsCommon,
) (ServicesGroupOut, error)

func NewService added in v1.21.0 

func NewService(app *fx.App, serviceName primitives.ServiceName, logger log.Logger) ServicesGroupOut

func WorkerServiceProvider added in v1.14.0 

func WorkerServiceProvider(
	params ServiceProviderParamsCommon,
) (ServicesGroupOut, error)

type ServicesMetadata added in v1.14.0 

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

func (*ServicesMetadata) Stop added in v1.21.0 

func (svc *ServicesMetadata) Stop(ctx context.Context)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.