Add cron schedule to customized search attribute

Author: neil-xie

common/definition/indexedKeys.go

@@ -40,6 +40,7 @@ const (
 	ClusterAttributeScope = "ClusterAttributeScope"
 	ClusterAttributeName  = "ClusterAttributeName"
 	IsCron                = "IsCron"
+	CronSchedule          = "CronSchedule"
 	NumClusters           = "NumClusters"
 	UpdateTime            = "UpdateTime"
 	CustomDomain          = "CustomDomain" // to support batch workflow
@@ -76,6 +77,7 @@ func createDefaultIndexedKeys() map[string]interface{} {
 		CustomDatetimeField:  types.IndexedValueTypeDatetime,
 		CadenceChangeVersion: types.IndexedValueTypeKeyword,
 		BinaryChecksums:      types.IndexedValueTypeKeyword,
+		CronSchedule:         types.IndexedValueTypeKeyword,
 		CustomDomain:         types.IndexedValueTypeString,
 		Operator:             types.IndexedValueTypeString,
 	}

config/dynamicconfig/development.yaml

@@ -66,6 +66,7 @@ frontend.validSearchAttributes:
     service: 1
     user: 1
     IsDeleted: 4
+    CronSchedule: 1
   constraints: {}
 system.writeVisibilityStoreName:
   - value: "db"

config/dynamicconfig/development_es.yaml

@@ -40,6 +40,7 @@ frontend.validSearchAttributes:
       BinaryChecksums: 1
       Passed: 4
       ShardID: 2
+      CronSchedule: 1
 system.minRetentionDays:
     - value: 0
 history.EnableConsistentQueryByDomain:

config/dynamicconfig/development_pinot.yaml

@@ -41,6 +41,7 @@ frontend.validSearchAttributes:
       Passed: 4
       ShardID: 2
       IsDeleted: 4
+      CronSchedule: 1
 system.minRetentionDays:
     - value: 0
 history.EnableConsistentQueryByDomain:

schema/elasticsearch/os2/visibility/index_template.json

@@ -61,6 +61,9 @@
           },
           "user": {
             "type": "keyword"
+          },
+          "CronSchedule": {
+            "type": "keyword"
           }
         }
       },

schema/elasticsearch/v6/visibility/index_template.json

@@ -83,7 +83,8 @@
             "Operator": { "type": "keyword"},
             "RolloutID": { "type": "keyword"},
             "BinaryChecksums": { "type": "keyword"},
-            "Passed": { "type": "boolean" }
+            "Passed": { "type": "boolean" },
+            "CronSchedule": { "type": "keyword" }
           }
         }
       }

schema/elasticsearch/v7/visibility/index_template.json

@@ -82,7 +82,8 @@
           "Operator": { "type": "keyword"},
           "RolloutID": { "type": "keyword"},
           "BinaryChecksums": { "type": "keyword"},
-          "Passed": { "type": "boolean" }
+          "Passed": { "type": "boolean" },
+          "CronSchedule": { "type": "keyword" }
         }
       }
     }

service/history/task/transfer_active_task_executor.go

@@ -23,6 +23,7 @@ package task
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"time"
@@ -35,6 +36,7 @@ import (
 	"github.com/uber/cadence/common/cache"
 	"github.com/uber/cadence/common/clock"
 	"github.com/uber/cadence/common/constants"
+	"github.com/uber/cadence/common/definition"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/metrics"
@@ -442,6 +444,18 @@ func (t *transferActiveTaskExecutor) processCloseExecutionTaskHelper(
 	domainName := mutableState.GetDomainEntry().GetInfo().Name
 	children := mutableState.GetPendingChildExecutionInfos()
 
+	// Add CronSchedule to search attributes if it's a cron workflow
+	if isCron {
+		searchAttr = copySearchAttributes(searchAttr)
+		if searchAttr == nil {
+			searchAttr = make(map[string][]byte)
+		}
+		cronScheduleBytes, err := json.Marshal(executionInfo.CronSchedule)
+		if err == nil {
+			searchAttr[definition.CronSchedule] = cronScheduleBytes
+		}
+	}
+
 	// we've gathered all necessary information from mutable state.
 	// release the context lock since we no longer need mutable state builder and
 	// the rest of logic is making RPC call, which takes time.
@@ -1007,6 +1021,17 @@ func (t *transferActiveTaskExecutor) processRecordWorkflowStartedOrUpsertHelper(
 	numClusters := (int16)(len(domainEntry.GetReplicationConfig().Clusters))
 	updateTimestamp := t.shard.GetTimeSource().Now()
 
+	// Add CronSchedule to search attributes if it's a cron workflow
+	if isCron {
+		if searchAttr == nil {
+			searchAttr = make(map[string][]byte)
+		}
+		cronScheduleBytes, err := json.Marshal(executionInfo.CronSchedule)
+		if err == nil {
+			searchAttr[definition.CronSchedule] = cronScheduleBytes
+		}
+	}
+
 	// release the context lock since we no longer need mutable state builder and
 	// the rest of logic is making RPC call, which takes time.
 	release(nil)

service/history/task/transfer_standby_task_executor.go

@@ -22,10 +22,12 @@ package task
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"time"
 
 	"github.com/uber/cadence/common/constants"
+	"github.com/uber/cadence/common/definition"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/metrics"
@@ -286,6 +288,18 @@ func (t *transferStandbyTaskExecutor) processCloseExecution(
 		isCron := len(executionInfo.CronSchedule) > 0
 		updateTimestamp := t.shard.GetTimeSource().Now()
 
+		// Add CronSchedule to search attributes if it's a cron workflow
+		if isCron {
+			searchAttr = copySearchAttributes(searchAttr)
+			if searchAttr == nil {
+				searchAttr = make(map[string][]byte)
+			}
+			cronScheduleBytes, err := json.Marshal(executionInfo.CronSchedule)
+			if err == nil {
+				searchAttr[definition.CronSchedule] = cronScheduleBytes
+			}
+		}
+
 		lastWriteVersion, err := mutableState.GetLastWriteVersion()
 		if err != nil {
 			return nil, err
@@ -537,6 +551,17 @@ func (t *transferStandbyTaskExecutor) processRecordWorkflowStartedOrUpsertHelper
 	searchAttr := copySearchAttributes(executionInfo.SearchAttributes)
 	headers := getWorkflowHeaders(startEvent)
 
+	// Add CronSchedule to search attributes if it's a cron workflow
+	if isCron {
+		if searchAttr == nil {
+			searchAttr = make(map[string][]byte)
+		}
+		cronScheduleBytes, err := json.Marshal(executionInfo.CronSchedule)
+		if err == nil {
+			searchAttr[definition.CronSchedule] = cronScheduleBytes
+		}
+	}
+
 	if isRecordStart {
 		workflowStartedScope.IncCounter(metrics.WorkflowStartedCount)
 		return t.recordWorkflowStarted(