Add GetIndexName method to visibility manager (#3820)

temporalio · Jan 27, 2023 · dc257ce · dc257ce
1 parent fc6efef
commit dc257ce
Show file tree

Hide file tree

Showing 31 changed files with 253 additions and 108 deletions.
diff --git a/common/persistence/sql/common.go b/common/persistence/sql/common.go
@@ -57,6 +57,10 @@ func (m *SqlStore) GetName() string {
 	return m.Db.PluginName()
 }
 
+func (m *SqlStore) GetDbName() string {
+	return m.Db.DbName()
+}
+
 func (m *SqlStore) Close() {
 	if m.Db != nil {
 		err := m.Db.Close()

diff --git a/common/persistence/sql/sqlplugin/interfaces.go b/common/persistence/sql/sqlplugin/interfaces.go
@@ -110,6 +110,7 @@ type (
 
 		BeginTx(ctx context.Context) (Tx, error)
 		PluginName() string
+		DbName() string
 		IsDupEntryError(err error) bool
 		Close() error
 	}

diff --git a/common/persistence/sql/sqlplugin/mysql/db.go b/common/persistence/sql/sqlplugin/mysql/db.go
@@ -111,6 +111,11 @@ func (mdb *db) PluginName() string {
 	return PluginName
 }
 
+// DbName returns the name of the database
+func (mdb *db) DbName() string {
+	return mdb.dbName
+}
+
 // ExpectedVersion returns expected version.
 func (mdb *db) ExpectedVersion() string {
 	switch mdb.dbKind {

diff --git a/common/persistence/sql/sqlplugin/postgresql/db.go b/common/persistence/sql/sqlplugin/postgresql/db.go
@@ -110,6 +110,11 @@ func (pdb *db) PluginName() string {
 	return PluginName
 }
 
+// DbName returns the name of the database
+func (pdb *db) DbName() string {
+	return pdb.dbName
+}
+
 // ExpectedVersion returns expected version.
 func (pdb *db) ExpectedVersion() string {
 	switch pdb.dbKind {

diff --git a/common/persistence/sql/sqlplugin/sqlite/db.go b/common/persistence/sql/sqlplugin/sqlite/db.go
@@ -122,6 +122,11 @@ func (mdb *db) PluginName() string {
 	return PluginName
 }
 
+// DbName returns the name of the database
+func (mdb *db) DbName() string {
+	return mdb.dbName
+}
+
 // ExpectedVersion returns expected version.
 func (mdb *db) ExpectedVersion() string {
 	switch mdb.dbKind {

diff --git a/common/persistence/visibility/manager/visibility_manager.go b/common/persistence/visibility/manager/visibility_manager.go
@@ -44,6 +44,7 @@ type (
 	VisibilityManager interface {
 		persistence.Closeable
 		GetName() string
+		GetIndexName() string
 
 		// Write APIs.
 		RecordWorkflowExecutionStarted(ctx context.Context, request *RecordWorkflowExecutionStartedRequest) error

diff --git a/common/persistence/visibility/manager/visibility_manager_mock.go b/common/persistence/visibility/manager/visibility_manager_mock.go
diff --git a/common/persistence/visibility/store/elasticsearch/visibility_store.go b/common/persistence/visibility/store/elasticsearch/visibility_store.go
@@ -130,6 +130,10 @@ func (s *visibilityStore) GetName() string {
 	return PersistenceName
 }
 
+func (s *visibilityStore) GetIndexName() string {
+	return s.index
+}
+
 func (s *visibilityStore) RecordWorkflowExecutionStarted(
 	ctx context.Context,
 	request *store.InternalRecordWorkflowExecutionStartedRequest,

diff --git a/common/persistence/visibility/store/standard/cassandra/visibility_store.go b/common/persistence/visibility/store/standard/cassandra/visibility_store.go
@@ -140,6 +140,7 @@ type (
 	visibilityStore struct {
 		session      gocql.Session
 		lowConslevel gocql.Consistency
+		keyspace     string
 	}
 )
 
@@ -158,13 +159,18 @@ func NewVisibilityStore(
 	return &visibilityStore{
 		session:      session,
 		lowConslevel: gocql.One,
+		keyspace:     cfg.Keyspace,
 	}, nil
 }
 
 func (v *visibilityStore) GetName() string {
 	return CassandraPersistenceName
 }
 
+func (v *visibilityStore) GetIndexName() string {
+	return v.keyspace
+}
+
 // Close releases the resources held by this object
 func (v *visibilityStore) Close() {
 	v.session.Close()

diff --git a/common/persistence/visibility/store/standard/sql/visibility_store.go b/common/persistence/visibility/store/standard/sql/visibility_store.go
@@ -80,6 +80,10 @@ func (s *visibilityStore) GetName() string {
 	return s.sqlStore.GetName()
 }
 
+func (s *visibilityStore) GetIndexName() string {
+	return s.sqlStore.GetDbName()
+}
+
 func (s *visibilityStore) RecordWorkflowExecutionStarted(
 	ctx context.Context,
 	request *store.InternalRecordWorkflowExecutionStartedRequest,

diff --git a/common/persistence/visibility/store/standard/visibility_store.go b/common/persistence/visibility/store/standard/visibility_store.go
@@ -73,6 +73,10 @@ func (s *standardStore) GetName() string {
 	return s.store.GetName()
 }
 
+func (s *standardStore) GetIndexName() string {
+	return s.store.GetIndexName()
+}
+
 func (s *standardStore) RecordWorkflowExecutionStarted(
 	ctx context.Context,
 	request *store.InternalRecordWorkflowExecutionStartedRequest,

diff --git a/common/persistence/visibility/store/visibility_store.go b/common/persistence/visibility/store/visibility_store.go
@@ -43,6 +43,7 @@ type (
 	VisibilityStore interface {
 		persistence.Closeable
 		GetName() string
+		GetIndexName() string
 
 		// Write APIs.
 		RecordWorkflowExecutionStarted(ctx context.Context, request *InternalRecordWorkflowExecutionStartedRequest) error

diff --git a/common/persistence/visibility/store/visibility_store_mock.go b/common/persistence/visibility/store/visibility_store_mock.go
diff --git a/common/persistence/visibility/visibility_manager_dual.go b/common/persistence/visibility/visibility_manager_dual.go
@@ -64,6 +64,10 @@ func (v *visibilityManagerDual) GetName() string {
 	return strings.Join([]string{v.visibilityManager.GetName(), v.secondaryVisibilityManager.GetName()}, ",")
 }
 
+func (v *visibilityManagerDual) GetIndexName() string {
+	return v.visibilityManager.GetIndexName()
+}
+
 func (v *visibilityManagerDual) RecordWorkflowExecutionStarted(
 	ctx context.Context,
 	request *manager.RecordWorkflowExecutionStartedRequest,

diff --git a/common/persistence/visibility/visibility_manager_impl.go b/common/persistence/visibility/visibility_manager_impl.go
@@ -76,6 +76,10 @@ func (p *visibilityManagerImpl) GetName() string {
 	return p.store.GetName()
 }
 
+func (p *visibilityManagerImpl) GetIndexName() string {
+	return p.store.GetIndexName()
+}
+
 func (p *visibilityManagerImpl) RecordWorkflowExecutionStarted(
 	ctx context.Context,
 	request *manager.RecordWorkflowExecutionStartedRequest,

diff --git a/common/persistence/visibility/visibility_manager_rate_limited.go b/common/persistence/visibility/visibility_manager_rate_limited.go
@@ -67,6 +67,10 @@ func (m *visibilityManagerRateLimited) GetName() string {
 	return m.delegate.GetName()
 }
 
+func (m *visibilityManagerRateLimited) GetIndexName() string {
+	return m.delegate.GetIndexName()
+}
+
 // Below are write APIs.
 
 func (m *visibilityManagerRateLimited) RecordWorkflowExecutionStarted(

diff --git a/common/persistence/visibility/visiblity_manager_metrics.go b/common/persistence/visibility/visiblity_manager_metrics.go
@@ -69,6 +69,10 @@ func (m *visibilityManagerMetrics) GetName() string {
 	return m.delegate.GetName()
 }
 
+func (m *visibilityManagerMetrics) GetIndexName() string {
+	return m.delegate.GetIndexName()
+}
+
 func (m *visibilityManagerMetrics) RecordWorkflowExecutionStarted(
 	ctx context.Context,
 	request *manager.RecordWorkflowExecutionStartedRequest,

diff --git a/common/resourcetest/resourceTest.go b/common/resourcetest/resourceTest.go
@@ -51,6 +51,7 @@ import (
 	"go.temporal.io/server/common/persistence"
 	persistenceClient "go.temporal.io/server/common/persistence/client"
 	"go.temporal.io/server/common/persistence/serialization"
+	"go.temporal.io/server/common/persistence/visibility/manager"
 	esclient "go.temporal.io/server/common/persistence/visibility/store/elasticsearch/client"
 	"go.temporal.io/server/common/primitives"
 	"go.temporal.io/server/common/sdk"
@@ -96,6 +97,7 @@ type (
 		ClientBean           *client.MockBean
 		ClientFactory        *client.MockFactory
 		ESClient             *esclient.MockClient
+		VisibilityManager    *manager.MockVisibilityManager
 
 		// persistence clients
 
@@ -206,6 +208,7 @@ func NewTest(
 		ClientBean:           clientBean,
 		ClientFactory:        clientFactory,
 		ESClient:             esclient.NewMockClient(controller),
+		VisibilityManager:    manager.NewMockVisibilityManager(controller),
 
 		// persistence clients
 
@@ -373,6 +376,11 @@ func (t *Test) GetClientFactory() client.Factory {
 	return t.ClientFactory
 }
 
+// GetVisibilityManager for testing
+func (t *Test) GetVisibilityManager() manager.VisibilityManager {
+	return t.VisibilityManager
+}
+
 // persistence clients
 
 // GetMetadataManager for testing

diff --git a/common/searchattribute/manager.go b/common/searchattribute/manager.go
@@ -87,7 +87,7 @@ func NewManager(
 }
 
 // GetSearchAttributes returns all search attributes (including system and build-in) for specified index.
-// indexName can be an empty string when Elasticsearch is not configured.
+// indexName can be an empty string for backward compatibility.
 func (m *managerImpl) GetSearchAttributes(
 	indexName string,
 	forceRefreshCache bool,
@@ -111,11 +111,19 @@ func (m *managerImpl) GetSearchAttributes(
 	}
 
 	indexSearchAttributes, ok := saCache.searchAttributes[indexName]
-	if !ok {
-		return NameTypeMap{}, nil
+	if ok {
+		return indexSearchAttributes, nil
 	}
 
-	return indexSearchAttributes, nil
+	if indexName != "" {
+		// Try to look for the empty string indexName for backward compatibility: up to v1.19,
+		// empty string was used when Elasticsearch was not configured.
+		indexSearchAttributes, ok = saCache.searchAttributes[""]
+		if ok {
+			return indexSearchAttributes, nil
+		}
+	}
+	return NameTypeMap{}, nil
 }
 
 func (m *managerImpl) needRefreshCache(saCache cache, forceRefreshCache bool, now time.Time) bool {

diff --git a/service/frontend/adminHandler.go b/service/frontend/adminHandler.go
@@ -97,7 +97,6 @@ type (
 
 		logger                      log.Logger
 		numberOfHistoryShards       int32
-		ESConfig                    *esclient.Config
 		ESClient                    esclient.Client
 		config                      *Config
 		namespaceHandler            namespace.Handler
@@ -127,7 +126,6 @@ type (
 		Config                              *Config
 		NamespaceReplicationQueue           persistence.NamespaceReplicationQueue
 		ReplicatorNamespaceReplicationQueue persistence.NamespaceReplicationQueue
-		EsConfig                            *esclient.Config
 		EsClient                            esclient.Client
 		VisibilityMrg                       manager.VisibilityManager
 		Logger                              log.Logger
@@ -192,7 +190,6 @@ func NewAdminHandler(
 		),
 		eventSerializer:             args.EventSerializer,
 		visibilityMgr:               args.VisibilityMrg,
-		ESConfig:                    args.EsConfig,
 		ESClient:                    args.EsClient,
 		persistenceExecutionManager: args.PersistenceExecutionManager,
 		namespaceReplicationQueue:   args.NamespaceReplicationQueue,
@@ -257,7 +254,7 @@ func (adh *AdminHandler) AddSearchAttributes(ctx context.Context, request *admin
 
 	indexName := request.GetIndexName()
 	if indexName == "" {
-		indexName = adh.ESConfig.GetVisibilityIndex()
+		indexName = adh.visibilityMgr.GetIndexName()
 	}
 
 	currentSearchAttributes, err := adh.saProvider.GetSearchAttributes(indexName, true)
@@ -322,7 +319,7 @@ func (adh *AdminHandler) RemoveSearchAttributes(ctx context.Context, request *ad
 
 	indexName := request.GetIndexName()
 	if indexName == "" {
-		indexName = adh.ESConfig.GetVisibilityIndex()
+		indexName = adh.visibilityMgr.GetIndexName()
 	}
 
 	currentSearchAttributes, err := adh.saProvider.GetSearchAttributes(indexName, true)
@@ -359,7 +356,7 @@ func (adh *AdminHandler) GetSearchAttributes(ctx context.Context, request *admin
 
 	indexName := request.GetIndexName()
 	if indexName == "" {
-		indexName = adh.ESConfig.GetVisibilityIndex()
+		indexName = adh.visibilityMgr.GetIndexName()
 	}
 
 	resp, err := adh.getSearchAttributes(ctx, indexName, "")