Handle updating cluster metadata search attributes info when using SQL DB (#3861)

Author: rodrigozhou

common/config/persistence.go

@@ -81,6 +81,10 @@ func (c *Persistence) AdvancedVisibilityConfigExist() bool {
 	return c.AdvancedVisibilityStore != ""
 }
 
+func (c *Persistence) IsSQLVisibilityStore() bool {
+	return c.StandardVisibilityConfigExist() && c.DataStores[c.VisibilityStore].SQL != nil
+}
+
 func (c *Persistence) validateAdvancedVisibility() error {
 	if !c.StandardVisibilityConfigExist() && !c.AdvancedVisibilityConfigExist() {
 		return errors.New("persistence config: one of visibilityStore or advancedVisibilityStore must be specified")

common/searchattribute/defs.go

@@ -28,6 +28,7 @@ import (
 	"strings"
 
 	enumspb "go.temporal.io/api/enums/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
 )
 
 const (
@@ -102,6 +103,37 @@ var (
 		Memo:              {},
 		VisibilityTaskKey: {},
 	}
+
+	sqlDbCustomSearchAttributes = map[string]enumspb.IndexedValueType{
+		"Bool01":        enumspb.INDEXED_VALUE_TYPE_BOOL,
+		"Bool02":        enumspb.INDEXED_VALUE_TYPE_BOOL,
+		"Bool03":        enumspb.INDEXED_VALUE_TYPE_BOOL,
+		"Datetime01":    enumspb.INDEXED_VALUE_TYPE_DATETIME,
+		"Datetime02":    enumspb.INDEXED_VALUE_TYPE_DATETIME,
+		"Datetime03":    enumspb.INDEXED_VALUE_TYPE_DATETIME,
+		"Double01":      enumspb.INDEXED_VALUE_TYPE_DOUBLE,
+		"Double02":      enumspb.INDEXED_VALUE_TYPE_DOUBLE,
+		"Double03":      enumspb.INDEXED_VALUE_TYPE_DOUBLE,
+		"Int01":         enumspb.INDEXED_VALUE_TYPE_INT,
+		"Int02":         enumspb.INDEXED_VALUE_TYPE_INT,
+		"Int03":         enumspb.INDEXED_VALUE_TYPE_INT,
+		"Keyword01":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Keyword02":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Keyword03":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Keyword04":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Keyword05":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Keyword06":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Keyword07":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Keyword08":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Keyword09":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Keyword10":     enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		"Text01":        enumspb.INDEXED_VALUE_TYPE_TEXT,
+		"Text02":        enumspb.INDEXED_VALUE_TYPE_TEXT,
+		"Text03":        enumspb.INDEXED_VALUE_TYPE_TEXT,
+		"KeywordList01": enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST,
+		"KeywordList02": enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST,
+		"KeywordList03": enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST,
+	}
 )
 
 // IsReserved returns true if name is system reserved and can't be used as custom search attribute name.
@@ -128,3 +160,9 @@ func IsMappable(name string) bool {
 	}
 	return true
 }
+
+func GetSqlDbIndexSearchAttributes() *persistencespb.IndexSearchAttributes {
+	return &persistencespb.IndexSearchAttributes{
+		CustomSearchAttributes: sqlDbCustomSearchAttributes,
+	}
+}

temporal/fx.go

@@ -644,6 +644,15 @@ func ApplyClusterMetadataConfigProvider(
 	}
 	defer clusterMetadataManager.Close()
 
+	var indexName string
+	var initialIndexSearchAttributes map[string]*persistencespb.IndexSearchAttributes
+	if config.Persistence.IsSQLVisibilityStore() {
+		indexName = config.Persistence.DataStores[config.Persistence.VisibilityStore].SQL.DatabaseName
+		initialIndexSearchAttributes = map[string]*persistencespb.IndexSearchAttributes{
+			indexName: searchattribute.GetSqlDbIndexSearchAttributes(),
+		}
+	}
+
 	clusterData := config.ClusterMetadata
 	for clusterName, clusterInfo := range clusterData.ClusterInformation {
 		if clusterName != clusterData.CurrentClusterName {
@@ -671,6 +680,7 @@ func ApplyClusterMetadataConfigProvider(
 					IsGlobalNamespaceEnabled: clusterData.EnableGlobalNamespace,
 					IsConnectionEnabled:      clusterInfo.Enabled,
 					UseClusterIdMembership:   true, // Enable this for new cluster after 1.19. This is to prevent two clusters join into one ring.
+					IndexSearchAttributes:    initialIndexSearchAttributes,
 				},
 			})
 		if err != nil {
@@ -681,21 +691,51 @@ func ApplyClusterMetadataConfigProvider(
 			continue
 		}
 
-		resp, err := clusterMetadataManager.GetClusterMetadata(ctx, &persistence.GetClusterMetadataRequest{
-			ClusterName: clusterName,
-		})
+		resp, err := clusterMetadataManager.GetClusterMetadata(
+			ctx,
+			&persistence.GetClusterMetadataRequest{ClusterName: clusterName},
+		)
 		if err != nil {
 			return config.ClusterMetadata, config.Persistence, fmt.Errorf("error while fetching cluster metadata: %w", err)
 		}
 
+		currentMetadata := resp.ClusterMetadata
+
+		// TODO (rodrigozhou): Remove this block for v1.21.
+		// Handle registering custom search attributes when upgrading to v1.20.
+		if config.Persistence.IsSQLVisibilityStore() {
+			needSave := false
+			if currentMetadata.IndexSearchAttributes == nil {
+				currentMetadata.IndexSearchAttributes = initialIndexSearchAttributes
+				needSave = true
+			} else if _, ok := currentMetadata.IndexSearchAttributes[indexName]; !ok {
+				currentMetadata.IndexSearchAttributes[indexName] = searchattribute.GetSqlDbIndexSearchAttributes()
+				needSave = true
+			}
+
+			if needSave {
+				_, err := clusterMetadataManager.SaveClusterMetadata(
+					ctx,
+					&persistence.SaveClusterMetadataRequest{ClusterMetadata: currentMetadata},
+				)
+				if err != nil {
+					logger.Warn(
+						"Failed to register search attributes.",
+						tag.Error(err),
+						tag.ClusterName(clusterName),
+					)
+				}
+				logger.Info("Successfully registered search attributes.", tag.ClusterName(clusterName))
+			}
+		}
+
 		// Allow updating cluster metadata if global namespace is disabled
 		if !resp.IsGlobalNamespaceEnabled && clusterData.EnableGlobalNamespace {
-			currentMetadata := resp.ClusterMetadata
 			currentMetadata.IsGlobalNamespaceEnabled = clusterData.EnableGlobalNamespace
 			currentMetadata.InitialFailoverVersion = clusterInfo.InitialFailoverVersion
 			currentMetadata.FailoverVersionIncrement = clusterData.FailoverVersionIncrement
 
-			applied, err = clusterMetadataManager.SaveClusterMetadata(
+			applied, err := clusterMetadataManager.SaveClusterMetadata(
 				ctx,
 				&persistence.SaveClusterMetadataRequest{
 					ClusterMetadata: currentMetadata,