mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
Refactor onebox (#10984)
## What changed? Several purely refactoring changes to onebox; **only** moves code and unexports types and fields. ## Why? Prepare the https://github.com/temporalio/temporal/pull/10319 PR by refactoring a few things first.
This commit is contained in:
Stephan Behnke
@@ -4,6 +4,7 @@ import (
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"go.temporal.io/api/operatorservice/v1"
|
||||
"go.temporal.io/api/workflowservice/v1"
|
||||
@@ -11,23 +12,39 @@ import (
|
||||
"go.temporal.io/server/api/historyservice/v1"
|
||||
"go.temporal.io/server/api/matchingservice/v1"
|
||||
schedulerpb "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
|
||||
"go.temporal.io/server/client"
|
||||
matchingclient "go.temporal.io/server/client/matching"
|
||||
"go.temporal.io/server/common"
|
||||
"go.temporal.io/server/common/cluster"
|
||||
"go.temporal.io/server/common/config"
|
||||
"go.temporal.io/server/common/dynamicconfig"
|
||||
"go.temporal.io/server/common/log"
|
||||
"go.temporal.io/server/common/log/tag"
|
||||
"go.temporal.io/server/common/membership"
|
||||
"go.temporal.io/server/common/membership/static"
|
||||
"go.temporal.io/server/common/metrics"
|
||||
"go.temporal.io/server/common/namespace"
|
||||
"go.temporal.io/server/common/persistence"
|
||||
"go.temporal.io/server/common/primitives"
|
||||
"go.temporal.io/server/common/rpc"
|
||||
"go.temporal.io/server/common/rpc/auth"
|
||||
"go.temporal.io/server/common/rpc/encryption"
|
||||
"go.temporal.io/server/common/sdk"
|
||||
"go.temporal.io/server/common/testing/testhooks"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
type clients struct {
|
||||
logger log.Logger
|
||||
hostsByService map[primitives.ServiceName]static.Hosts
|
||||
tlsConfigProvider *encryption.FixedTLSConfigProvider
|
||||
// The matching client is built lazily because routing depends on the
|
||||
// frontend membership address populated during startup.
|
||||
newMatchingClient func() (matchingservice.MatchingServiceClient, error)
|
||||
logger log.Logger
|
||||
hostsByService map[primitives.ServiceName]static.Hosts
|
||||
frontendMembershipAddress *string
|
||||
tlsConfigProvider *encryption.FixedTLSConfigProvider
|
||||
metricsHandler metrics.Handler
|
||||
dcClient *dynamicconfig.MemoryClient
|
||||
testHooks testhooks.TestHooks
|
||||
numHistoryShards int32
|
||||
metadataMgr persistence.MetadataManager
|
||||
tokenProvider auth.TokenProvider
|
||||
|
||||
frontend frontendClients
|
||||
history historyClients
|
||||
@@ -57,14 +74,26 @@ type matchingClient struct {
|
||||
func newClients(
|
||||
logger log.Logger,
|
||||
hostsByService map[primitives.ServiceName]static.Hosts,
|
||||
frontendMembershipAddress *string,
|
||||
tlsConfigProvider *encryption.FixedTLSConfigProvider,
|
||||
newMatchingClient func() (matchingservice.MatchingServiceClient, error),
|
||||
metricsHandler metrics.Handler,
|
||||
dcClient *dynamicconfig.MemoryClient,
|
||||
testHooks testhooks.TestHooks,
|
||||
numHistoryShards int32,
|
||||
metadataMgr persistence.MetadataManager,
|
||||
tokenProvider auth.TokenProvider,
|
||||
) clients {
|
||||
return clients{
|
||||
logger: logger,
|
||||
hostsByService: hostsByService,
|
||||
tlsConfigProvider: tlsConfigProvider,
|
||||
newMatchingClient: newMatchingClient,
|
||||
logger: logger,
|
||||
hostsByService: hostsByService,
|
||||
frontendMembershipAddress: frontendMembershipAddress,
|
||||
tlsConfigProvider: tlsConfigProvider,
|
||||
metricsHandler: metricsHandler,
|
||||
dcClient: dcClient,
|
||||
testHooks: testHooks,
|
||||
numHistoryShards: numHistoryShards,
|
||||
metadataMgr: metadataMgr,
|
||||
tokenProvider: tokenProvider,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -125,11 +154,60 @@ func (c *clients) MatchingClient() matchingservice.MatchingServiceClient {
|
||||
|
||||
func (c *clients) ensureMatching() {
|
||||
c.matching.once.Do(func() {
|
||||
client, err := c.newMatchingClient()
|
||||
var tlsConfigProvider encryption.TLSConfigProvider
|
||||
var frontendTLSConfig *tls.Config
|
||||
if c.tlsConfigProvider != nil {
|
||||
var err error
|
||||
tlsConfigProvider = c.tlsConfigProvider
|
||||
frontendTLSConfig, err = c.tlsConfigProvider.GetFrontendClientConfig()
|
||||
if err != nil {
|
||||
c.logger.Fatal("failed getting client TLS config", tag.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
monitor := static.NewMonitor(c.hostsByService)
|
||||
monitor.Start()
|
||||
rpcFactory := rpc.NewFactory(
|
||||
&config.Config{},
|
||||
primitives.FrontendService,
|
||||
c.logger,
|
||||
c.metricsHandler,
|
||||
tlsConfigProvider,
|
||||
*c.frontendMembershipAddress,
|
||||
*c.frontendMembershipAddress,
|
||||
0,
|
||||
frontendTLSConfig,
|
||||
nil,
|
||||
nil,
|
||||
monitor,
|
||||
c.tokenProvider,
|
||||
)
|
||||
clientFactory := client.NewFactoryProvider().NewFactory(
|
||||
rpcFactory,
|
||||
monitor,
|
||||
c.metricsHandler,
|
||||
dynamicconfig.NewCollection(c.dcClient, c.logger),
|
||||
c.testHooks,
|
||||
c.numHistoryShards,
|
||||
c.logger,
|
||||
c.logger,
|
||||
)
|
||||
namespaceIDToName := func(id namespace.ID) (namespace.Name, error) {
|
||||
resp, err := c.metadataMgr.GetNamespace(NewContext(), &persistence.GetNamespaceRequest{ID: id.String()})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return namespace.Name(resp.Namespace.Info.Name), nil
|
||||
}
|
||||
matchingClient, err := clientFactory.NewMatchingClientWithTimeout(
|
||||
namespaceIDToName,
|
||||
matchingclient.DefaultTimeout,
|
||||
matchingclient.DefaultLongPollTimeout,
|
||||
)
|
||||
if err != nil {
|
||||
c.logger.Fatal("unable to create matching test client", tag.Error(err))
|
||||
}
|
||||
c.matching.client = client
|
||||
c.matching.client = matchingClient
|
||||
})
|
||||
}
|
||||
|
||||
@@ -178,3 +256,88 @@ func (c *clients) tlsConfig(serviceName primitives.ServiceName) (*tls.Config, er
|
||||
}
|
||||
return c.tlsConfigProvider.GetInternodeClientConfig()
|
||||
}
|
||||
|
||||
func newClientFactoryProvider(
|
||||
clusterConfig *cluster.Config,
|
||||
mockAdminClient map[string]adminservice.AdminServiceClient,
|
||||
) client.FactoryProvider {
|
||||
return &clientFactoryProvider{
|
||||
config: clusterConfig,
|
||||
mockAdminClient: mockAdminClient,
|
||||
}
|
||||
}
|
||||
|
||||
type clientFactoryProvider struct {
|
||||
config *cluster.Config
|
||||
mockAdminClient map[string]adminservice.AdminServiceClient
|
||||
}
|
||||
|
||||
func (p *clientFactoryProvider) NewFactory(
|
||||
rpcFactory common.RPCFactory,
|
||||
monitor membership.Monitor,
|
||||
metricsHandler metrics.Handler,
|
||||
dc *dynamicconfig.Collection,
|
||||
testHooks testhooks.TestHooks,
|
||||
numberOfHistoryShards int32,
|
||||
logger log.Logger,
|
||||
throttledLogger log.Logger,
|
||||
) client.Factory {
|
||||
f := client.NewFactoryProvider().NewFactory(
|
||||
rpcFactory,
|
||||
monitor,
|
||||
metricsHandler,
|
||||
dc,
|
||||
testHooks,
|
||||
numberOfHistoryShards,
|
||||
logger,
|
||||
throttledLogger,
|
||||
)
|
||||
return &clientFactory{
|
||||
Factory: f,
|
||||
config: p.config,
|
||||
mockAdminClient: p.mockAdminClient,
|
||||
}
|
||||
}
|
||||
|
||||
type clientFactory struct {
|
||||
client.Factory
|
||||
config *cluster.Config
|
||||
mockAdminClient map[string]adminservice.AdminServiceClient
|
||||
}
|
||||
|
||||
// override just this one and look up connections in mock admin client map
|
||||
func (f *clientFactory) NewRemoteAdminClientWithTimeout(rpcAddress string, timeout time.Duration, largeTimeout time.Duration) adminservice.AdminServiceClient {
|
||||
var clusterName string
|
||||
for name, info := range f.config.ClusterInformation {
|
||||
if rpcAddress == info.RPCAddress {
|
||||
clusterName = name
|
||||
}
|
||||
}
|
||||
if mock, ok := f.mockAdminClient[clusterName]; ok {
|
||||
return mock
|
||||
}
|
||||
return f.Factory.NewRemoteAdminClientWithTimeout(rpcAddress, timeout, largeTimeout)
|
||||
}
|
||||
|
||||
func sdkClientFactoryProvider(
|
||||
grpcResolver *membership.GRPCResolver,
|
||||
metricsHandler metrics.Handler,
|
||||
logger log.Logger,
|
||||
dc *dynamicconfig.Collection,
|
||||
tlsConfigProvider encryption.TLSConfigProvider,
|
||||
) sdk.ClientFactory {
|
||||
var tlsConfig *tls.Config
|
||||
if tlsConfigProvider != nil {
|
||||
var err error
|
||||
if tlsConfig, err = tlsConfigProvider.GetFrontendClientConfig(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
return sdk.NewClientFactory(
|
||||
grpcResolver.MakeURL(primitives.FrontendService),
|
||||
tlsConfig,
|
||||
metricsHandler,
|
||||
logger,
|
||||
dynamicconfig.WorkerStickyCacheSize.Get(dc),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -89,8 +89,8 @@ type (
|
||||
// and will panic if called.
|
||||
isShared bool
|
||||
}
|
||||
// TestClusterParams contains the variables which are used to configure test cluster via the TestClusterOption type.
|
||||
TestClusterParams struct {
|
||||
// testClusterParams contains the variables which are used to configure test cluster via the TestClusterOption type.
|
||||
testClusterParams struct {
|
||||
DCRedirectionPolicy config.DCRedirectionPolicy
|
||||
DynamicConfigOverrides map[dynamicconfig.Key]any
|
||||
ArchivalEnabled bool
|
||||
@@ -104,7 +104,7 @@ type (
|
||||
CustomHistoryArchiverFactory provider.CustomHistoryArchiverFactory
|
||||
CustomVisibilityArchiverFactory provider.CustomVisibilityArchiverFactory
|
||||
}
|
||||
TestClusterOption func(params *TestClusterParams)
|
||||
TestClusterOption func(params *testClusterParams)
|
||||
)
|
||||
|
||||
func init() {
|
||||
@@ -115,13 +115,13 @@ func init() {
|
||||
}
|
||||
|
||||
func WithDCRedirectionPolicy(policy config.DCRedirectionPolicy) TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.DCRedirectionPolicy = policy
|
||||
}
|
||||
}
|
||||
|
||||
func WithDynamicConfigOverrides(overrides map[dynamicconfig.Key]any) TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
if params.DynamicConfigOverrides == nil {
|
||||
params.DynamicConfigOverrides = overrides
|
||||
} else {
|
||||
@@ -131,31 +131,31 @@ func WithDynamicConfigOverrides(overrides map[dynamicconfig.Key]any) TestCluster
|
||||
}
|
||||
|
||||
func WithArchivalEnabled() TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.ArchivalEnabled = true
|
||||
}
|
||||
}
|
||||
|
||||
func withMTLS() TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.EnableMTLS = true
|
||||
}
|
||||
}
|
||||
|
||||
func withWorkerService(enabled bool) TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.EnableWorkerService = enabled
|
||||
}
|
||||
}
|
||||
|
||||
func WithFaultInjectionConfig(cfg *config.FaultInjection) TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.FaultInjectionConfig = cfg
|
||||
}
|
||||
}
|
||||
|
||||
func WithNumHistoryShards(n int32) TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.NumHistoryShards = n
|
||||
}
|
||||
}
|
||||
@@ -163,31 +163,31 @@ func WithNumHistoryShards(n int32) TestClusterOption {
|
||||
// WithClusterLogger sets a custom logger for the test cluster, used instead of
|
||||
// the default test logger. Useful for intercepting server log output.
|
||||
func WithClusterLogger(logger log.Logger) TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.Logger = logger
|
||||
}
|
||||
}
|
||||
|
||||
func WithClusterHistoryTaskRecorder() TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.EnableHistoryTaskRecorder = true
|
||||
}
|
||||
}
|
||||
|
||||
func WithSharedCluster() TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.SharedCluster = true
|
||||
}
|
||||
}
|
||||
|
||||
func WithCustomHistoryArchiverFactory(factory provider.CustomHistoryArchiverFactory) TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.CustomHistoryArchiverFactory = factory
|
||||
}
|
||||
}
|
||||
|
||||
func WithCustomVisibilityArchiverFactory(factory provider.CustomVisibilityArchiverFactory) TestClusterOption {
|
||||
return func(params *TestClusterParams) {
|
||||
return func(params *testClusterParams) {
|
||||
params.CustomVisibilityArchiverFactory = factory
|
||||
}
|
||||
}
|
||||
@@ -386,8 +386,8 @@ func (s *FunctionalTestBase) checkTestShard() {
|
||||
checkTestShard(s.T())
|
||||
}
|
||||
|
||||
func ApplyTestClusterOptions(options []TestClusterOption) TestClusterParams {
|
||||
params := TestClusterParams{
|
||||
func ApplyTestClusterOptions(options []TestClusterOption) testClusterParams {
|
||||
params := testClusterParams{
|
||||
EnableWorkerService: true,
|
||||
}
|
||||
for _, opt := range options {
|
||||
|
||||
@@ -17,11 +17,8 @@ import (
|
||||
|
||||
sdktrace "go.opentelemetry.io/otel/sdk/trace"
|
||||
"go.temporal.io/server/api/adminservice/v1"
|
||||
"go.temporal.io/server/api/matchingservice/v1"
|
||||
"go.temporal.io/server/chasm"
|
||||
chasmnexus "go.temporal.io/server/chasm/lib/nexusoperation"
|
||||
"go.temporal.io/server/client"
|
||||
matchingclient "go.temporal.io/server/client/matching"
|
||||
"go.temporal.io/server/common"
|
||||
carchiver "go.temporal.io/server/common/archiver"
|
||||
"go.temporal.io/server/common/archiver/provider"
|
||||
@@ -47,7 +44,6 @@ import (
|
||||
"go.temporal.io/server/common/rpc"
|
||||
"go.temporal.io/server/common/rpc/auth"
|
||||
"go.temporal.io/server/common/rpc/encryption"
|
||||
"go.temporal.io/server/common/sdk"
|
||||
"go.temporal.io/server/common/searchattribute"
|
||||
"go.temporal.io/server/common/telemetry"
|
||||
"go.temporal.io/server/common/testing/testhooks"
|
||||
@@ -65,7 +61,7 @@ import (
|
||||
)
|
||||
|
||||
type (
|
||||
TemporalImpl struct {
|
||||
temporalImpl struct {
|
||||
clients
|
||||
fxApps []*fx.App
|
||||
|
||||
@@ -136,41 +132,41 @@ type (
|
||||
DisableWorker bool // overrides NumWorkers
|
||||
}
|
||||
|
||||
// TemporalParams contains everything needed to bootstrap Temporal
|
||||
TemporalParams struct {
|
||||
ClusterMetadataConfig *cluster.Config
|
||||
PersistenceConfig config.Persistence
|
||||
MetadataMgr persistence.MetadataManager
|
||||
ClusterMetadataManager persistence.ClusterMetadataManager
|
||||
ShardMgr persistence.ShardManager
|
||||
ExecutionManager persistence.ExecutionManager
|
||||
TaskMgr persistence.TaskManager
|
||||
NamespaceReplicationQueue persistence.NamespaceReplicationQueue
|
||||
AbstractDataStoreFactory persistenceClient.AbstractDataStoreFactory
|
||||
VisibilityStoreFactory visibility.VisibilityStoreFactory
|
||||
Logger log.Logger
|
||||
ArchiverMetadata carchiver.ArchivalMetadata
|
||||
ArchiverProvider provider.ArchiverProvider
|
||||
EnableReadHistoryFromArchival bool
|
||||
FrontendConfig FrontendConfig
|
||||
HistoryConfig HistoryConfig
|
||||
MatchingConfig MatchingConfig
|
||||
WorkerConfig WorkerConfig
|
||||
ESConfig *esclient.Config
|
||||
ESClient esclient.Client
|
||||
MockAdminClient map[string]adminservice.AdminServiceClient
|
||||
NamespaceReplicationTaskExecutor nsreplication.TaskExecutor
|
||||
DCRedirectionPolicy config.DCRedirectionPolicy
|
||||
DynamicConfigOverrides map[dynamicconfig.Key]any
|
||||
TLSConfigProvider *encryption.FixedTLSConfigProvider
|
||||
CaptureMetricsHandler *metricstest.CaptureHandler
|
||||
// temporalParams contains everything needed to bootstrap Temporal
|
||||
temporalParams struct {
|
||||
clusterMetadataConfig *cluster.Config
|
||||
persistenceConfig config.Persistence
|
||||
metadataMgr persistence.MetadataManager
|
||||
clusterMetadataManager persistence.ClusterMetadataManager
|
||||
shardMgr persistence.ShardManager
|
||||
executionManager persistence.ExecutionManager
|
||||
taskMgr persistence.TaskManager
|
||||
namespaceReplicationQueue persistence.NamespaceReplicationQueue
|
||||
abstractDataStoreFactory persistenceClient.AbstractDataStoreFactory
|
||||
visibilityStoreFactory visibility.VisibilityStoreFactory
|
||||
logger log.Logger
|
||||
archiverMetadata carchiver.ArchivalMetadata
|
||||
archiverProvider provider.ArchiverProvider
|
||||
enableReadHistoryFromArchival bool
|
||||
frontendConfig FrontendConfig
|
||||
historyConfig HistoryConfig
|
||||
matchingConfig MatchingConfig
|
||||
workerConfig WorkerConfig
|
||||
esConfig *esclient.Config
|
||||
esClient esclient.Client
|
||||
mockAdminClient map[string]adminservice.AdminServiceClient
|
||||
namespaceReplicationTaskExecutor nsreplication.TaskExecutor
|
||||
dcRedirectionPolicy config.DCRedirectionPolicy
|
||||
dynamicConfigOverrides map[dynamicconfig.Key]any
|
||||
tlsConfigProvider *encryption.FixedTLSConfigProvider
|
||||
captureMetricsHandler *metricstest.CaptureHandler
|
||||
// ServiceFxOptions is populated by WithFxOptionsForService.
|
||||
ServiceFxOptions map[primitives.ServiceName][]fx.Option
|
||||
TaskCategoryRegistry tasks.TaskCategoryRegistry
|
||||
HostsByProtocolByService map[transferProtocol]map[primitives.ServiceName]static.Hosts
|
||||
SpanExporters map[telemetry.SpanExporterType]sdktrace.SpanExporter
|
||||
TokenProvider auth.TokenProvider
|
||||
EnableHistoryTaskRecorder bool
|
||||
serviceFxOptions map[primitives.ServiceName][]fx.Option
|
||||
taskCategoryRegistry tasks.TaskCategoryRegistry
|
||||
hostsByProtocolByService map[transferProtocol]map[primitives.ServiceName]static.Hosts
|
||||
spanExporters map[telemetry.SpanExporterType]sdktrace.SpanExporter
|
||||
tokenProvider auth.TokenProvider
|
||||
enableHistoryTaskRecorder bool
|
||||
}
|
||||
|
||||
listenHostPort string
|
||||
@@ -180,52 +176,58 @@ type (
|
||||
const NamespaceCacheRefreshInterval = time.Second
|
||||
|
||||
// newTemporal returns an instance that hosts full temporal in one process
|
||||
func newTemporal(t *testing.T, params *TemporalParams) *TemporalImpl {
|
||||
impl := &TemporalImpl{
|
||||
logger: params.Logger,
|
||||
clusterMetadataConfig: params.ClusterMetadataConfig,
|
||||
persistenceConfig: params.PersistenceConfig,
|
||||
metadataMgr: params.MetadataMgr,
|
||||
clusterMetadataMgr: params.ClusterMetadataManager,
|
||||
shardMgr: params.ShardMgr,
|
||||
taskMgr: params.TaskMgr,
|
||||
executionManager: params.ExecutionManager,
|
||||
namespaceReplicationQueue: params.NamespaceReplicationQueue,
|
||||
abstractDataStoreFactory: params.AbstractDataStoreFactory,
|
||||
visibilityStoreFactory: params.VisibilityStoreFactory,
|
||||
esConfig: params.ESConfig,
|
||||
esClient: params.ESClient,
|
||||
archiverMetadata: params.ArchiverMetadata,
|
||||
archiverProvider: params.ArchiverProvider,
|
||||
frontendConfig: params.FrontendConfig,
|
||||
historyConfig: params.HistoryConfig,
|
||||
matchingConfig: params.MatchingConfig,
|
||||
workerConfig: params.WorkerConfig,
|
||||
mockAdminClient: params.MockAdminClient,
|
||||
namespaceReplicationTaskExecutor: params.NamespaceReplicationTaskExecutor,
|
||||
dcRedirectionPolicy: params.DCRedirectionPolicy,
|
||||
tlsConfigProvider: params.TLSConfigProvider,
|
||||
captureMetricsHandler: params.CaptureMetricsHandler,
|
||||
func newTemporal(t *testing.T, params *temporalParams) *temporalImpl {
|
||||
impl := &temporalImpl{
|
||||
logger: params.logger,
|
||||
clusterMetadataConfig: params.clusterMetadataConfig,
|
||||
persistenceConfig: params.persistenceConfig,
|
||||
metadataMgr: params.metadataMgr,
|
||||
clusterMetadataMgr: params.clusterMetadataManager,
|
||||
shardMgr: params.shardMgr,
|
||||
taskMgr: params.taskMgr,
|
||||
executionManager: params.executionManager,
|
||||
namespaceReplicationQueue: params.namespaceReplicationQueue,
|
||||
abstractDataStoreFactory: params.abstractDataStoreFactory,
|
||||
visibilityStoreFactory: params.visibilityStoreFactory,
|
||||
esConfig: params.esConfig,
|
||||
esClient: params.esClient,
|
||||
archiverMetadata: params.archiverMetadata,
|
||||
archiverProvider: params.archiverProvider,
|
||||
frontendConfig: params.frontendConfig,
|
||||
historyConfig: params.historyConfig,
|
||||
matchingConfig: params.matchingConfig,
|
||||
workerConfig: params.workerConfig,
|
||||
mockAdminClient: params.mockAdminClient,
|
||||
namespaceReplicationTaskExecutor: params.namespaceReplicationTaskExecutor,
|
||||
dcRedirectionPolicy: params.dcRedirectionPolicy,
|
||||
tlsConfigProvider: params.tlsConfigProvider,
|
||||
captureMetricsHandler: params.captureMetricsHandler,
|
||||
dcClient: dynamicconfig.NewMemoryClient(),
|
||||
testHooks: testhooks.NewTestHooks(),
|
||||
serviceFxOptions: params.ServiceFxOptions,
|
||||
taskCategoryRegistry: params.TaskCategoryRegistry,
|
||||
hostsByProtocolByService: params.HostsByProtocolByService,
|
||||
serviceFxOptions: params.serviceFxOptions,
|
||||
taskCategoryRegistry: params.taskCategoryRegistry,
|
||||
hostsByProtocolByService: params.hostsByProtocolByService,
|
||||
replicationStreamRecorder: NewReplicationStreamRecorder(),
|
||||
spanExporters: params.SpanExporters,
|
||||
tokenProvider: params.TokenProvider,
|
||||
enableHistoryTaskRecorder: params.EnableHistoryTaskRecorder,
|
||||
spanExporters: params.spanExporters,
|
||||
tokenProvider: params.tokenProvider,
|
||||
enableHistoryTaskRecorder: params.enableHistoryTaskRecorder,
|
||||
}
|
||||
|
||||
// Configure output file path for on-demand logging (call WriteToLog() to write)
|
||||
clusterName := params.ClusterMetadataConfig.CurrentClusterName
|
||||
clusterName := params.clusterMetadataConfig.CurrentClusterName
|
||||
outputFile := fmt.Sprintf("/tmp/replication_stream_messages_%s.txt", clusterName)
|
||||
impl.replicationStreamRecorder.SetOutputFile(outputFile)
|
||||
impl.clients = newClients(
|
||||
impl.logger,
|
||||
impl.hostsByProtocolByService[grpcProtocol],
|
||||
&impl.frontendMembershipAddress,
|
||||
impl.tlsConfigProvider,
|
||||
impl.newMatchingClient,
|
||||
impl.GetMetricsHandler(),
|
||||
impl.dcClient,
|
||||
impl.testHooks,
|
||||
impl.historyConfig.NumHistoryShards,
|
||||
impl.metadataMgr,
|
||||
impl.tokenProvider,
|
||||
)
|
||||
|
||||
// Global defaults: applied without cleanup so they persist across cluster reuse.
|
||||
@@ -236,66 +238,13 @@ func newTemporal(t *testing.T, params *TemporalParams) *TemporalImpl {
|
||||
// so it can't be overriden in the loop above.
|
||||
impl.setNexusCallbackURL()
|
||||
// Per-test overrides: cleaned up when the creating test finishes.
|
||||
for k, v := range params.DynamicConfigOverrides {
|
||||
for k, v := range params.dynamicConfigOverrides {
|
||||
impl.overrideDynamicConfigForTest(t, k, v)
|
||||
}
|
||||
return impl
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) newMatchingClient() (matchingservice.MatchingServiceClient, error) {
|
||||
var tlsConfigProvider encryption.TLSConfigProvider
|
||||
var frontendTLSConfig *tls.Config
|
||||
if c.tlsConfigProvider != nil {
|
||||
var err error
|
||||
tlsConfigProvider = c.tlsConfigProvider
|
||||
frontendTLSConfig, err = c.tlsConfigProvider.GetFrontendClientConfig()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed getting client TLS config: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
monitor := static.NewMonitor(c.hostsByProtocolByService[grpcProtocol])
|
||||
monitor.Start()
|
||||
rpcFactory := rpc.NewFactory(
|
||||
&config.Config{},
|
||||
primitives.FrontendService,
|
||||
c.logger,
|
||||
c.GetMetricsHandler(),
|
||||
tlsConfigProvider,
|
||||
c.frontendMembershipAddress,
|
||||
c.frontendMembershipAddress,
|
||||
0,
|
||||
frontendTLSConfig,
|
||||
nil,
|
||||
nil,
|
||||
monitor,
|
||||
c.tokenProvider,
|
||||
)
|
||||
clientFactory := client.NewFactoryProvider().NewFactory(
|
||||
rpcFactory,
|
||||
monitor,
|
||||
c.GetMetricsHandler(),
|
||||
dynamicconfig.NewCollection(c.dcClient, c.logger),
|
||||
c.testHooks,
|
||||
c.historyConfig.NumHistoryShards,
|
||||
c.logger,
|
||||
c.logger,
|
||||
)
|
||||
namespaceIDToName := func(id namespace.ID) (namespace.Name, error) {
|
||||
resp, err := c.metadataMgr.GetNamespace(NewContext(), &persistence.GetNamespaceRequest{ID: id.String()})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return namespace.Name(resp.Namespace.Info.Name), nil
|
||||
}
|
||||
return clientFactory.NewMatchingClientWithTimeout(
|
||||
namespaceIDToName,
|
||||
matchingclient.DefaultTimeout,
|
||||
matchingclient.DefaultLongPollTimeout,
|
||||
)
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) Start() error {
|
||||
func (c *temporalImpl) Start() error {
|
||||
// create temporal-system namespace, this must be created before starting
|
||||
// the services - so directly use the metadataManager to create this
|
||||
if err := c.createSystemNamespace(); err != nil {
|
||||
@@ -309,7 +258,7 @@ func (c *TemporalImpl) Start() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) Stop() error {
|
||||
func (c *temporalImpl) Stop() error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -324,7 +273,7 @@ func (c *TemporalImpl) Stop() error {
|
||||
return multierr.Combine(errs...)
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) makeHostMap(serviceName primitives.ServiceName, self string) map[primitives.ServiceName]static.Hosts {
|
||||
func (c *temporalImpl) makeHostMap(serviceName primitives.ServiceName, self string) map[primitives.ServiceName]static.Hosts {
|
||||
hostMap := maps.Clone(c.hostsByProtocolByService[grpcProtocol])
|
||||
hosts := hostMap[serviceName]
|
||||
hosts.Self = self
|
||||
@@ -333,29 +282,29 @@ func (c *TemporalImpl) makeHostMap(serviceName primitives.ServiceName, self stri
|
||||
}
|
||||
|
||||
// Use this to get an address for a remote cluster to connect to.
|
||||
func (c *TemporalImpl) RemoteFrontendGRPCAddress() string {
|
||||
func (c *temporalImpl) RemoteFrontendGRPCAddress() string {
|
||||
return c.hostsByProtocolByService[grpcProtocol][primitives.FrontendService].All[0]
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) FrontendHTTPAddress() string {
|
||||
func (c *temporalImpl) FrontendHTTPAddress() string {
|
||||
// randomize like a load balancer would
|
||||
addrs := c.hostsByProtocolByService[httpProtocol][primitives.FrontendService].All
|
||||
return addrs[rand.Intn(len(addrs))]
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) FrontendGRPCAddress() string {
|
||||
func (c *temporalImpl) FrontendGRPCAddress() string {
|
||||
return c.hostsByProtocolByService[grpcProtocol][primitives.FrontendService].All[0]
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) WorkerGRPCAddress() string {
|
||||
func (c *temporalImpl) WorkerGRPCAddress() string {
|
||||
return c.hostsByProtocolByService[grpcProtocol][primitives.WorkerService].All[0]
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) DcClient() *dynamicconfig.MemoryClient {
|
||||
func (c *temporalImpl) DcClient() *dynamicconfig.MemoryClient {
|
||||
return c.dcClient
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) ChasmContext(ctx context.Context) (context.Context, error) {
|
||||
func (c *temporalImpl) ChasmContext(ctx context.Context) (context.Context, error) {
|
||||
if numHistoryHosts := len(c.hostsByProtocolByService[grpcProtocol][primitives.HistoryService].All); numHistoryHosts != 1 {
|
||||
return nil, fmt.Errorf("expected exactly one history host for chasm context, got %d", numHistoryHosts)
|
||||
}
|
||||
@@ -367,7 +316,7 @@ func (c *TemporalImpl) ChasmContext(ctx context.Context) (context.Context, error
|
||||
return ctx, nil
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) copyPersistenceConfig() config.Persistence {
|
||||
func (c *temporalImpl) copyPersistenceConfig() config.Persistence {
|
||||
persistenceConfig := copyPersistenceConfig(c.persistenceConfig)
|
||||
if c.esConfig != nil {
|
||||
esDataStoreName := "es-visibility"
|
||||
@@ -379,7 +328,7 @@ func (c *TemporalImpl) copyPersistenceConfig() config.Persistence {
|
||||
return persistenceConfig
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) startFrontend() {
|
||||
func (c *temporalImpl) startFrontend() {
|
||||
serviceName := primitives.FrontendService
|
||||
|
||||
var grpcResolver *membership.GRPCResolver
|
||||
@@ -425,7 +374,7 @@ func (c *TemporalImpl) startFrontend() {
|
||||
fx.Provide(func() authorization.Authorizer { return c }),
|
||||
fx.Provide(func() authorization.ClaimMapper { return c }),
|
||||
fx.Provide(func() authorization.JWTAudienceMapper { return nil }),
|
||||
fx.Provide(c.newClientFactoryProvider),
|
||||
fx.Provide(newClientFactoryProvider),
|
||||
fx.Provide(func() searchattribute.Mapper { return nil }),
|
||||
// Comment the line above and uncomment the line below to test with search attributes mapper.
|
||||
// fx.Provide(func() searchattribute.Mapper { return NewSearchAttributeTestMapper() }),
|
||||
@@ -463,7 +412,7 @@ func (c *TemporalImpl) startFrontend() {
|
||||
c.frontendMembershipAddress = grpcResolver.MakeURL(serviceName)
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) startHistory() {
|
||||
func (c *temporalImpl) startHistory() {
|
||||
serviceName := primitives.HistoryService
|
||||
|
||||
testhooks.NewHook(testhooks.HistoryChasmRuntimeProvider, func(
|
||||
@@ -524,7 +473,7 @@ func (c *TemporalImpl) startHistory() {
|
||||
fx.Provide(func() carchiver.ArchivalMetadata { return c.archiverMetadata }),
|
||||
fx.Provide(func() provider.ArchiverProvider { return c.archiverProvider }),
|
||||
fx.Provide(sdkClientFactoryProvider),
|
||||
fx.Provide(c.newClientFactoryProvider),
|
||||
fx.Provide(newClientFactoryProvider),
|
||||
fx.Provide(func() searchattribute.Mapper { return nil }),
|
||||
// Comment the line above and uncomment the line below to test with search attributes mapper.
|
||||
// fx.Provide(func() searchattribute.Mapper { return NewSearchAttributeTestMapper() }),
|
||||
@@ -560,7 +509,7 @@ func (c *TemporalImpl) startHistory() {
|
||||
}
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) startMatching() {
|
||||
func (c *temporalImpl) startMatching() {
|
||||
serviceName := primitives.MatchingService
|
||||
|
||||
for _, host := range c.hostsByProtocolByService[grpcProtocol][serviceName].All {
|
||||
@@ -583,7 +532,7 @@ func (c *TemporalImpl) startMatching() {
|
||||
fx.Provide(func() *cluster.Config { return c.clusterMetadataConfig }),
|
||||
fx.Provide(func() carchiver.ArchivalMetadata { return c.archiverMetadata }),
|
||||
fx.Provide(func() provider.ArchiverProvider { return c.archiverProvider }),
|
||||
fx.Provide(c.newClientFactoryProvider),
|
||||
fx.Provide(newClientFactoryProvider),
|
||||
fx.Provide(func() searchattribute.Mapper { return nil }),
|
||||
fx.Provide(func() resolver.ServiceResolver { return resolver.NewNoopResolver() }),
|
||||
fx.Provide(persistenceClient.FactoryProvider),
|
||||
@@ -614,7 +563,7 @@ func (c *TemporalImpl) startMatching() {
|
||||
}
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) startWorker() {
|
||||
func (c *temporalImpl) startWorker() {
|
||||
serviceName := primitives.WorkerService
|
||||
|
||||
clusterConfigCopy := cluster.Config{
|
||||
@@ -648,7 +597,7 @@ func (c *TemporalImpl) startWorker() {
|
||||
fx.Provide(func() carchiver.ArchivalMetadata { return c.archiverMetadata }),
|
||||
fx.Provide(func() provider.ArchiverProvider { return c.archiverProvider }),
|
||||
fx.Provide(sdkClientFactoryProvider),
|
||||
fx.Provide(c.newClientFactoryProvider),
|
||||
fx.Provide(newClientFactoryProvider),
|
||||
fx.Provide(func() searchattribute.Mapper { return nil }),
|
||||
fx.Provide(func() resolver.ServiceResolver { return resolver.NewNoopResolver() }),
|
||||
fx.Provide(persistenceClient.FactoryProvider),
|
||||
@@ -680,11 +629,11 @@ func (c *TemporalImpl) startWorker() {
|
||||
}
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) getFxOptionsForService(serviceName primitives.ServiceName) fx.Option {
|
||||
func (c *temporalImpl) getFxOptionsForService(serviceName primitives.ServiceName) fx.Option {
|
||||
return fx.Options(c.serviceFxOptions[serviceName]...)
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) createSystemNamespace() error {
|
||||
func (c *temporalImpl) createSystemNamespace() error {
|
||||
err := c.metadataMgr.InitializeSystemNamespaces(context.Background(), c.clusterMetadataConfig.CurrentClusterName)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create temporal-system namespace: %v", err)
|
||||
@@ -692,11 +641,11 @@ func (c *TemporalImpl) createSystemNamespace() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) GetHistoryTaskRecorder() *HistoryTaskRecorder {
|
||||
func (c *temporalImpl) GetHistoryTaskRecorder() *HistoryTaskRecorder {
|
||||
return c.historyTaskRecorder
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) GetTLSConfigProvider() encryption.TLSConfigProvider {
|
||||
func (c *temporalImpl) GetTLSConfigProvider() encryption.TLSConfigProvider {
|
||||
// If we just return this directly, the interface will be non-nil but the
|
||||
// pointer will be nil
|
||||
if c.tlsConfigProvider != nil {
|
||||
@@ -705,28 +654,28 @@ func (c *TemporalImpl) GetTLSConfigProvider() encryption.TLSConfigProvider {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) GetTaskCategoryRegistry() tasks.TaskCategoryRegistry {
|
||||
func (c *temporalImpl) GetTaskCategoryRegistry() tasks.TaskCategoryRegistry {
|
||||
return c.taskCategoryRegistry
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) TlsConfigProvider() *encryption.FixedTLSConfigProvider {
|
||||
func (c *temporalImpl) TLSConfigProvider() *encryption.FixedTLSConfigProvider {
|
||||
return c.tlsConfigProvider
|
||||
}
|
||||
|
||||
// Deprecated: metric capture is cluster-global.
|
||||
// Use (*TestEnv).StartGlobalMetricCapture() or (*TestEnv).StartNamespaceMetricCapture() instead.
|
||||
func (c *TemporalImpl) CaptureMetricsHandler() *metricstest.CaptureHandler {
|
||||
func (c *temporalImpl) CaptureMetricsHandler() *metricstest.CaptureHandler {
|
||||
return c.captureMetricsHandler
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) GetMetricsHandler() metrics.Handler {
|
||||
func (c *temporalImpl) GetMetricsHandler() metrics.Handler {
|
||||
if c.captureMetricsHandler != nil {
|
||||
return c.captureMetricsHandler
|
||||
}
|
||||
return metrics.NoopMetricsHandler
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) frontendConfigProvider() *config.Config {
|
||||
func (c *temporalImpl) frontendConfigProvider() *config.Config {
|
||||
// Set HTTP port and a test HTTP forwarded header
|
||||
return &config.Config{
|
||||
Services: map[string]config.Service{
|
||||
@@ -747,7 +696,7 @@ func (c *TemporalImpl) frontendConfigProvider() *config.Config {
|
||||
}
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) configProvider(serviceName primitives.ServiceName) *config.Config {
|
||||
func (c *temporalImpl) configProvider(serviceName primitives.ServiceName) *config.Config {
|
||||
return &config.Config{
|
||||
Services: map[string]config.Service{
|
||||
string(serviceName): {
|
||||
@@ -761,7 +710,7 @@ func (c *TemporalImpl) configProvider(serviceName primitives.ServiceName) *confi
|
||||
}
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) newRPCFactory(
|
||||
func (c *temporalImpl) newRPCFactory(
|
||||
sn primitives.ServiceName,
|
||||
grpcHostPort listenHostPort,
|
||||
logger log.Logger,
|
||||
@@ -822,75 +771,13 @@ func (c *TemporalImpl) newRPCFactory(
|
||||
), nil
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) newClientFactoryProvider(
|
||||
config *cluster.Config,
|
||||
mockAdminClient map[string]adminservice.AdminServiceClient,
|
||||
) client.FactoryProvider {
|
||||
return &clientFactoryProvider{
|
||||
config: config,
|
||||
mockAdminClient: mockAdminClient,
|
||||
}
|
||||
}
|
||||
|
||||
type clientFactoryProvider struct {
|
||||
config *cluster.Config
|
||||
mockAdminClient map[string]adminservice.AdminServiceClient
|
||||
}
|
||||
|
||||
func (p *clientFactoryProvider) NewFactory(
|
||||
rpcFactory common.RPCFactory,
|
||||
monitor membership.Monitor,
|
||||
metricsHandler metrics.Handler,
|
||||
dc *dynamicconfig.Collection,
|
||||
testHooks testhooks.TestHooks,
|
||||
numberOfHistoryShards int32,
|
||||
logger log.Logger,
|
||||
throttledLogger log.Logger,
|
||||
) client.Factory {
|
||||
f := client.NewFactoryProvider().NewFactory(
|
||||
rpcFactory,
|
||||
monitor,
|
||||
metricsHandler,
|
||||
dc,
|
||||
testHooks,
|
||||
numberOfHistoryShards,
|
||||
logger,
|
||||
throttledLogger,
|
||||
)
|
||||
return &clientFactory{
|
||||
Factory: f,
|
||||
config: p.config,
|
||||
mockAdminClient: p.mockAdminClient,
|
||||
}
|
||||
}
|
||||
|
||||
type clientFactory struct {
|
||||
client.Factory
|
||||
config *cluster.Config
|
||||
mockAdminClient map[string]adminservice.AdminServiceClient
|
||||
}
|
||||
|
||||
// override just this one and look up connections in mock admin client map
|
||||
func (f *clientFactory) NewRemoteAdminClientWithTimeout(rpcAddress string, timeout time.Duration, largeTimeout time.Duration) adminservice.AdminServiceClient {
|
||||
var clusterName string
|
||||
for name, info := range f.config.ClusterInformation {
|
||||
if rpcAddress == info.RPCAddress {
|
||||
clusterName = name
|
||||
}
|
||||
}
|
||||
if mock, ok := f.mockAdminClient[clusterName]; ok {
|
||||
return mock
|
||||
}
|
||||
return f.Factory.NewRemoteAdminClientWithTimeout(rpcAddress, timeout, largeTimeout)
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) SetOnGetClaims(fn func(*authorization.AuthInfo) (*authorization.Claims, error)) {
|
||||
func (c *temporalImpl) SetOnGetClaims(fn func(*authorization.AuthInfo) (*authorization.Claims, error)) {
|
||||
c.callbackLock.Lock()
|
||||
c.onGetClaims = fn
|
||||
c.callbackLock.Unlock()
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) GetClaims(authInfo *authorization.AuthInfo) (*authorization.Claims, error) {
|
||||
func (c *temporalImpl) GetClaims(authInfo *authorization.AuthInfo) (*authorization.Claims, error) {
|
||||
c.callbackLock.RLock()
|
||||
onGetClaims := c.onGetClaims
|
||||
c.callbackLock.RUnlock()
|
||||
@@ -900,7 +787,7 @@ func (c *TemporalImpl) GetClaims(authInfo *authorization.AuthInfo) (*authorizati
|
||||
return &authorization.Claims{System: authorization.RoleAdmin}, nil
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) SetOnAuthorize(
|
||||
func (c *temporalImpl) SetOnAuthorize(
|
||||
fn func(context.Context, *authorization.Claims, *authorization.CallTarget) (authorization.Result, error),
|
||||
) {
|
||||
c.callbackLock.Lock()
|
||||
@@ -908,7 +795,7 @@ func (c *TemporalImpl) SetOnAuthorize(
|
||||
c.callbackLock.Unlock()
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) Authorize(
|
||||
func (c *temporalImpl) Authorize(
|
||||
ctx context.Context,
|
||||
caller *authorization.Claims,
|
||||
target *authorization.CallTarget,
|
||||
@@ -951,30 +838,7 @@ func copyPersistenceConfig(cfg config.Persistence) config.Persistence {
|
||||
return newCfg
|
||||
}
|
||||
|
||||
func sdkClientFactoryProvider(
|
||||
grpcResolver *membership.GRPCResolver,
|
||||
metricsHandler metrics.Handler,
|
||||
logger log.Logger,
|
||||
dc *dynamicconfig.Collection,
|
||||
tlsConfigProvider encryption.TLSConfigProvider,
|
||||
) sdk.ClientFactory {
|
||||
var tlsConfig *tls.Config
|
||||
if tlsConfigProvider != nil {
|
||||
var err error
|
||||
if tlsConfig, err = tlsConfigProvider.GetFrontendClientConfig(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
return sdk.NewClientFactory(
|
||||
grpcResolver.MakeURL(primitives.FrontendService),
|
||||
tlsConfig,
|
||||
metricsHandler,
|
||||
logger,
|
||||
dynamicconfig.WorkerStickyCacheSize.Get(dc),
|
||||
)
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) setNexusCallbackURL() {
|
||||
func (c *temporalImpl) setNexusCallbackURL() {
|
||||
// Set Nexus callback URL with the cluster's HTTP address. This is a sensible default to avoid
|
||||
// users to need to manually set this.
|
||||
//nolint:revive // test callback endpoints are served by the local HTTP API in functional tests
|
||||
@@ -986,18 +850,18 @@ func (c *TemporalImpl) setNexusCallbackURL() {
|
||||
c.overrideDynamicConfigForClusterLifetime(chasmnexus.CallbackURLTemplate.Key(), nexusCallbackTemplate)
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) overrideDynamicConfigForClusterLifetime(name dynamicconfig.Key, value any) {
|
||||
func (c *temporalImpl) overrideDynamicConfigForClusterLifetime(name dynamicconfig.Key, value any) {
|
||||
c.dcClient.PartialOverrideValue(name, value)
|
||||
}
|
||||
|
||||
// overrideDynamicConfigForTest overrides a dynamic config value for the duration of the test.
|
||||
func (c *TemporalImpl) overrideDynamicConfigForTest(t *testing.T, name dynamicconfig.Key, value any) func() {
|
||||
func (c *temporalImpl) overrideDynamicConfigForTest(t *testing.T, name dynamicconfig.Key, value any) func() {
|
||||
cleanup := c.dcClient.PartialOverrideValue(name, value)
|
||||
t.Cleanup(cleanup)
|
||||
return cleanup
|
||||
}
|
||||
|
||||
func (c *TemporalImpl) injectHook(t *testing.T, hook testhooks.Hook, scope any) func() {
|
||||
func (c *temporalImpl) injectHook(t *testing.T, hook testhooks.Hook, scope any) func() {
|
||||
cleanup := hook.Apply(c.testHooks, scope)
|
||||
t.Cleanup(cleanup)
|
||||
return cleanup
|
||||
|
||||
@@ -59,7 +59,7 @@ type (
|
||||
// TestCluster is a testcore struct for functional tests
|
||||
TestCluster struct {
|
||||
testBase *persistencetests.TestBase
|
||||
host *TemporalImpl
|
||||
host *temporalImpl
|
||||
}
|
||||
|
||||
// TestClusterConfig are config for a test cluster
|
||||
@@ -94,7 +94,7 @@ type (
|
||||
}
|
||||
|
||||
defaultTestClusterFactory struct {
|
||||
tbFactory PersistenceTestBaseFactory
|
||||
tbFactory persistenceTestBaseFactory
|
||||
}
|
||||
)
|
||||
|
||||
@@ -109,16 +109,16 @@ func (f *defaultTestClusterFactory) NewCluster(t *testing.T, clusterConfig *Test
|
||||
|
||||
func NewTestClusterFactory() TestClusterFactory {
|
||||
tbFactory := &defaultPersistenceTestBaseFactory{}
|
||||
return NewTestClusterFactoryWithCustomTestBaseFactory(tbFactory)
|
||||
return newTestClusterFactoryWithCustomTestBaseFactory(tbFactory)
|
||||
}
|
||||
|
||||
func NewTestClusterFactoryWithCustomTestBaseFactory(tbFactory PersistenceTestBaseFactory) TestClusterFactory {
|
||||
func newTestClusterFactoryWithCustomTestBaseFactory(tbFactory persistenceTestBaseFactory) TestClusterFactory {
|
||||
return &defaultTestClusterFactory{
|
||||
tbFactory: tbFactory,
|
||||
}
|
||||
}
|
||||
|
||||
type PersistenceTestBaseFactory interface {
|
||||
type persistenceTestBaseFactory interface {
|
||||
NewTestBase(options *persistencetests.TestBaseOptions) *persistencetests.TestBase
|
||||
}
|
||||
|
||||
@@ -148,7 +148,7 @@ func newClusterWithPersistenceTestBaseFactory(
|
||||
t *testing.T,
|
||||
clusterConfig *TestClusterConfig,
|
||||
logger log.Logger,
|
||||
tbFactory PersistenceTestBaseFactory,
|
||||
tbFactory persistenceTestBaseFactory,
|
||||
) (*TestCluster, error) {
|
||||
// determine number of hosts per service
|
||||
const minNodes = 1
|
||||
@@ -298,41 +298,41 @@ func newClusterWithPersistenceTestBaseFactory(
|
||||
}
|
||||
}
|
||||
|
||||
temporalParams := &TemporalParams{
|
||||
ClusterMetadataConfig: clusterMetadataConfig,
|
||||
PersistenceConfig: pConfig,
|
||||
MetadataMgr: testBase.MetadataManager,
|
||||
ClusterMetadataManager: testBase.ClusterMetadataManager,
|
||||
ShardMgr: testBase.ShardMgr,
|
||||
ExecutionManager: testBase.ExecutionManager,
|
||||
NamespaceReplicationQueue: testBase.NamespaceReplicationQueue,
|
||||
AbstractDataStoreFactory: testBase.AbstractDataStoreFactory,
|
||||
VisibilityStoreFactory: testBase.VisibilityStoreFactory,
|
||||
TaskMgr: testBase.TaskMgr,
|
||||
Logger: logger,
|
||||
ESConfig: clusterConfig.ESConfig,
|
||||
ESClient: esClient,
|
||||
ArchiverMetadata: archiverMetadata,
|
||||
ArchiverProvider: archiverProvider,
|
||||
FrontendConfig: clusterConfig.FrontendConfig,
|
||||
HistoryConfig: clusterConfig.HistoryConfig,
|
||||
MatchingConfig: clusterConfig.MatchingConfig,
|
||||
WorkerConfig: clusterConfig.WorkerConfig,
|
||||
MockAdminClient: clusterConfig.MockAdminClient,
|
||||
NamespaceReplicationTaskExecutor: nsreplication.NewTaskExecutor(clusterConfig.ClusterMetadata.CurrentClusterName, testBase.MetadataManager, nsreplication.NewNoopDataMerger(), nsreplication.NewDefaultAdmitter(), logger, testhooks.TestHooks{}),
|
||||
DCRedirectionPolicy: clusterConfig.DCRedirectionPolicy,
|
||||
DynamicConfigOverrides: clusterConfig.DynamicConfigOverrides,
|
||||
TLSConfigProvider: tlsConfigProvider,
|
||||
ServiceFxOptions: clusterConfig.ServiceFxOptions,
|
||||
TaskCategoryRegistry: temporal.TaskCategoryRegistryProvider(archiverMetadata),
|
||||
HostsByProtocolByService: hostsByProtocolByService,
|
||||
SpanExporters: clusterConfig.SpanExporters,
|
||||
TokenProvider: clusterConfig.TokenProvider,
|
||||
EnableHistoryTaskRecorder: clusterConfig.EnableHistoryTaskRecorder,
|
||||
temporalParams := &temporalParams{
|
||||
clusterMetadataConfig: clusterMetadataConfig,
|
||||
persistenceConfig: pConfig,
|
||||
metadataMgr: testBase.MetadataManager,
|
||||
clusterMetadataManager: testBase.ClusterMetadataManager,
|
||||
shardMgr: testBase.ShardMgr,
|
||||
executionManager: testBase.ExecutionManager,
|
||||
namespaceReplicationQueue: testBase.NamespaceReplicationQueue,
|
||||
abstractDataStoreFactory: testBase.AbstractDataStoreFactory,
|
||||
visibilityStoreFactory: testBase.VisibilityStoreFactory,
|
||||
taskMgr: testBase.TaskMgr,
|
||||
logger: logger,
|
||||
esConfig: clusterConfig.ESConfig,
|
||||
esClient: esClient,
|
||||
archiverMetadata: archiverMetadata,
|
||||
archiverProvider: archiverProvider,
|
||||
frontendConfig: clusterConfig.FrontendConfig,
|
||||
historyConfig: clusterConfig.HistoryConfig,
|
||||
matchingConfig: clusterConfig.MatchingConfig,
|
||||
workerConfig: clusterConfig.WorkerConfig,
|
||||
mockAdminClient: clusterConfig.MockAdminClient,
|
||||
namespaceReplicationTaskExecutor: nsreplication.NewTaskExecutor(clusterConfig.ClusterMetadata.CurrentClusterName, testBase.MetadataManager, nsreplication.NewNoopDataMerger(), nsreplication.NewDefaultAdmitter(), logger, testhooks.TestHooks{}),
|
||||
dcRedirectionPolicy: clusterConfig.DCRedirectionPolicy,
|
||||
dynamicConfigOverrides: clusterConfig.DynamicConfigOverrides,
|
||||
tlsConfigProvider: tlsConfigProvider,
|
||||
serviceFxOptions: clusterConfig.ServiceFxOptions,
|
||||
taskCategoryRegistry: temporal.TaskCategoryRegistryProvider(archiverMetadata),
|
||||
hostsByProtocolByService: hostsByProtocolByService,
|
||||
spanExporters: clusterConfig.SpanExporters,
|
||||
tokenProvider: clusterConfig.TokenProvider,
|
||||
enableHistoryTaskRecorder: clusterConfig.EnableHistoryTaskRecorder,
|
||||
}
|
||||
|
||||
if clusterConfig.EnableMetricsCapture {
|
||||
temporalParams.CaptureMetricsHandler = metricstest.NewCaptureHandler()
|
||||
temporalParams.captureMetricsHandler = metricstest.NewCaptureHandler()
|
||||
}
|
||||
|
||||
err = newPProfInitializerImpl(logger, PprofTestPort).Start()
|
||||
@@ -553,7 +553,7 @@ func (tc *TestCluster) ExecutionManager() persistence.ExecutionManager {
|
||||
}
|
||||
|
||||
// TODO (alex): expose only needed objects from TemporalImpl.
|
||||
func (tc *TestCluster) Host() *TemporalImpl {
|
||||
func (tc *TestCluster) Host() *temporalImpl {
|
||||
return tc.host
|
||||
}
|
||||
|
||||
|
||||
@@ -58,7 +58,7 @@ func (s *TLSFunctionalSuite) TestHTTPMTLS() {
|
||||
// Create HTTP client with TLS config
|
||||
httpClient := http.Client{
|
||||
Transport: &http.Transport{
|
||||
TLSClientConfig: env.GetTestCluster().Host().TlsConfigProvider().FrontendClientConfig,
|
||||
TLSClientConfig: env.GetTestCluster().Host().TLSConfigProvider().FrontendClientConfig,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user