Add update time to ES visibility requests and records (#CDNC-2567)

common/definition/indexedKeys.go

@@ -39,6 +39,7 @@ const (
 	TaskList        = "TaskList"
 	IsCron          = "IsCron"
 	NumClusters     = "NumClusters"
+	UpdateTime      = "UpdateTime"
 	CustomDomain    = "CustomDomain" // to support batch workflow
 	Operator        = "Operator"     // to support batch workflow
 
@@ -100,6 +101,7 @@ var systemIndexedKeys = map[string]interface{}{
 	TaskList:      shared.IndexedValueTypeKeyword,
 	IsCron:        shared.IndexedValueTypeBool,
 	NumClusters:   shared.IndexedValueTypeInt,
+	UpdateTime:    shared.IndexedValueTypeInt,
 }
 
 // IsSystemIndexedKey return true is key is system added

common/elasticsearch/client_v6.go

@@ -676,6 +676,9 @@ func (c *elasticV6) convertSearchResultToVisibilityRecord(hit *elastic.SearchHit
 		NumClusters:      source.NumClusters,
 		SearchAttributes: source.Attr,
 	}
+	if source.UpdateTime != 0 {
+		record.UpdateTime = time.Unix(0, source.UpdateTime)
+	}
 	if source.CloseTime != 0 {
 		record.CloseTime = time.Unix(0, source.CloseTime)
 		record.Status = thrift.ToWorkflowExecutionCloseStatus(&source.CloseStatus)

common/elasticsearch/client_v7.go

@@ -669,6 +669,9 @@ func (c *elasticV7) convertSearchResultToVisibilityRecord(hit *elastic.SearchHit
 		NumClusters:      source.NumClusters,
 		SearchAttributes: source.Attr,
 	}
+	if source.UpdateTime != 0 {
+		record.UpdateTime = time.Unix(0, source.UpdateTime)
+	}
 	if source.CloseTime != 0 {
 		record.CloseTime = time.Unix(0, source.CloseTime)
 		record.Status = thrift.ToWorkflowExecutionCloseStatus(&source.CloseStatus)

common/elasticsearch/defs.go

@@ -39,6 +39,7 @@ const (
 	IsCron              = "IsCron"
 	NumClusters         = "NumClusters"
 	VisibilityOperation = "VisibilityOperation"
+	UpdateTime          = "UpdateTime"
 
 	KafkaKey = "KafkaKey"
 )

common/elasticsearch/interfaces.go

@@ -233,6 +233,7 @@ type (
 		TaskList      string
 		IsCron        bool
 		NumClusters   int16
+		UpdateTime    int64
 		Attr          map[string]interface{}
 	}
 

common/persistence/dataStoreInterfaces.go

@@ -644,6 +644,7 @@ type (
 		TaskList         string
 		IsCron           bool
 		NumClusters      int16
+		UpdateTime       time.Time
 		SearchAttributes map[string]interface{}
 	}
 
@@ -720,6 +721,7 @@ type (
 		RetentionPeriod    time.Duration
 		IsCron             bool
 		NumClusters        int16
+		UpdateTimestamp    time.Time
 	}
 
 	// InternalRecordWorkflowExecutionUninitializedRequest is used to add a record of a newly uninitialized execution
@@ -744,6 +746,7 @@ type (
 		TaskList           string
 		IsCron             bool
 		NumClusters        int16
+		UpdateTimestamp    time.Time
 		SearchAttributes   map[string][]byte
 	}
 

common/persistence/dataVisibilityManagerInterfaces.go

@@ -72,6 +72,7 @@ type (
 		TaskList           string
 		IsCron             bool
 		NumClusters        int16
+		UpdateTimestamp    int64
 		SearchAttributes   map[string][]byte
 	}
 
@@ -97,6 +98,7 @@ type (
 		TaskList           string
 		IsCron             bool
 		NumClusters        int16
+		UpdateTimestamp    int64
 		SearchAttributes   map[string][]byte
 	}
 

common/persistence/elasticsearch/esVisibilityStore.go

@@ -110,6 +110,7 @@ func (v *esVisibilityStore) RecordWorkflowExecutionStarted(
 		0, // will not be used
 		0, // will not be used
 		0, // will not be used
+		0, // will be updated when workflow execution updates
 	)
 	return v.producer.Publish(ctx, msg)
 }
@@ -137,6 +138,7 @@ func (v *esVisibilityStore) RecordWorkflowExecutionClosed(
 		request.CloseTimestamp.UnixNano(),
 		*thrift.FromWorkflowExecutionCloseStatus(&request.Status),
 		request.HistoryLength,
+		request.UpdateTimestamp.UnixNano(),
 	)
 	return v.producer.Publish(ctx, msg)
 }
@@ -178,6 +180,7 @@ func (v *esVisibilityStore) UpsertWorkflowExecution(
 		0, // will not be used
 		0, // will not be used
 		0, // will not be used
+		request.UpdateTimestamp.UnixNano(),
 	)
 	return v.producer.Publish(ctx, msg)
 }
@@ -749,6 +752,7 @@ func createVisibilityMessage(
 	endTimeUnixNano int64, // close execution
 	closeStatus workflow.WorkflowExecutionCloseStatus, // close execution
 	historyLength int64, // close execution
+	updateTimeUnixNano int64, // update execution
 ) *indexer.Message {
 	msgType := indexer.MessageTypeIndex
 
@@ -759,6 +763,7 @@ func createVisibilityMessage(
 		es.TaskList:      {Type: &es.FieldTypeString, StringData: common.StringPtr(taskList)},
 		es.IsCron:        {Type: &es.FieldTypeBool, BoolData: common.BoolPtr(isCron)},
 		es.NumClusters:   {Type: &es.FieldTypeInt, IntData: common.Int64Ptr(int64(NumClusters))},
+		es.UpdateTime:    {Type: &es.FieldTypeInt, IntData: common.Int64Ptr(updateTimeUnixNano)},
 	}
 
 	if len(memo) != 0 {

common/persistence/elasticsearch/esVisibilityStore_test.go

@@ -187,6 +187,7 @@ func (s *ESVisibilitySuite) TestRecordWorkflowExecutionClosed() {
 	request.HistoryLength = int64(20)
 	request.IsCron = false
 	request.NumClusters = 2
+	request.UpdateTimestamp = time.Unix(0, int64(213))
 	s.mockProducer.On("Publish", mock.Anything, mock.MatchedBy(func(input *indexer.Message) bool {
 		fields := input.Fields
 		s.Equal(request.DomainUUID, input.GetDomainID())
@@ -204,6 +205,7 @@ func (s *ESVisibilitySuite) TestRecordWorkflowExecutionClosed() {
 		s.Equal(request.IsCron, fields[es.IsCron].GetBoolData())
 		s.Equal((int64)(request.NumClusters), fields[es.NumClusters].GetIntData())
 		s.Equal(indexer.VisibilityOperationRecordClosed, *input.VisibilityOperation)
+		s.Equal(request.UpdateTimestamp.UnixNano(), fields[es.UpdateTime].GetIntData())
 		return true
 	})).Return(nil).Once()
 

common/persistence/visibilitySingleManager.go

@@ -104,6 +104,7 @@ func (v *visibilityManagerImpl) RecordWorkflowExecutionClosed(
 		RetentionPeriod:    common.SecondsToDuration(request.RetentionSeconds),
 		IsCron:             request.IsCron,
 		NumClusters:        request.NumClusters,
+		UpdateTimestamp:    time.Unix(0, request.UpdateTimestamp),
 	}
 	return v.persistence.RecordWorkflowExecutionClosed(ctx, req)
 }
@@ -137,6 +138,7 @@ func (v *visibilityManagerImpl) UpsertWorkflowExecution(
 		TaskList:           request.TaskList,
 		IsCron:             request.IsCron,
 		NumClusters:        request.NumClusters,
+		UpdateTimestamp:    time.Unix(0, request.UpdateTimestamp),
 		SearchAttributes:   request.SearchAttributes,
 	}
 	return v.persistence.UpsertWorkflowExecution(ctx, req)

common/types/mapper/thrift/shared.go

@@ -6266,6 +6266,7 @@ func FromWorkflowExecutionInfo(t *types.WorkflowExecutionInfo) *shared.WorkflowE
 		AutoResetPoints:  FromResetPoints(t.AutoResetPoints),
 		TaskList:         &t.TaskList,
 		IsCron:           &t.IsCron,
+		UpdateTime:       t.UpdateTime,
 	}
 }
 
@@ -6289,6 +6290,7 @@ func ToWorkflowExecutionInfo(t *shared.WorkflowExecutionInfo) *types.WorkflowExe
 		AutoResetPoints:  ToResetPoints(t.AutoResetPoints),
 		TaskList:         t.GetTaskList(),
 		IsCron:           t.GetIsCron(),
+		UpdateTime:       t.UpdateTime,
 	}
 }
 

common/types/shared.go

@@ -6891,6 +6891,7 @@ type WorkflowExecutionInfo struct {
 	AutoResetPoints   *ResetPoints                  `json:"autoResetPoints,omitempty"`
 	TaskList          string                        `json:"taskList,omitempty"`
 	IsCron            bool                          `json:"isCron,omitempty"`
+	UpdateTime        *int64                        `json:"updateTime,omitempty"`
 }
 
 // GetExecution is an internal getter (TBD...)
@@ -6941,6 +6942,14 @@ func (v *WorkflowExecutionInfo) GetExecutionTime() (o int64) {
 	return
 }
 
+// GetUpdateTime is an internal getter (TBD...)
+func (v *WorkflowExecutionInfo) GetUpdateTime() (o int64) {
+	if v != nil && v.UpdateTime != nil {
+		return *v.UpdateTime
+	}
+	return
+}
+
 // GetSearchAttributes is an internal getter (TBD...)
 func (v *WorkflowExecutionInfo) GetSearchAttributes() (o *SearchAttributes) {
 	if v != nil && v.SearchAttributes != nil {

config/dynamicconfig/development_es.yaml

@@ -18,6 +18,7 @@ frontend.validSearchAttributes:
       TaskList: 1
       IsCron: 1
       NumClusters: 2
+      UpdateTime: 2
       CustomStringField: 0
       CustomKeywordField: 1
       CustomIntField: 2

schema/elasticsearch/v6/visibility/index_template.json

@@ -52,6 +52,9 @@
         "NumClusters": {
           "type": "integer"
         },
+        "UpdateTime": {
+          "type": "long"
+        },
         "Attr": {
           "properties": {
             "CadenceChangeVersion":  { "type": "keyword" },

schema/elasticsearch/v7/visibility/index_template.json

@@ -51,6 +51,9 @@
       "NumClusters": {
         "type": "integer"
       },
+      "UpdateTime": {
+        "type": "long"
+      },
       "Attr": {
         "properties": {
           "CadenceChangeVersion":  { "type": "keyword" },

service/history/historyEngine.go

@@ -1589,6 +1589,7 @@ func (e *historyEngineImpl) DescribeWorkflowExecution(
 			AutoResetPoints:  executionInfo.AutoResetPoints,
 			Memo:             &types.Memo{Fields: executionInfo.Memo},
 			IsCron:           len(executionInfo.CronSchedule) > 0,
+			UpdateTime:       common.Int64Ptr(executionInfo.LastUpdatedTimestamp.UnixNano()),
 			SearchAttributes: &types.SearchAttributes{IndexedFields: executionInfo.SearchAttributes},
 		},
 	}

service/history/task/transfer_active_task_executor.go

@@ -374,6 +374,7 @@ func (t *transferActiveTaskExecutor) processCloseExecutionTaskHelper(
 	workflowHistoryLength := mutableState.GetNextEventID() - 1
 	isCron := len(executionInfo.CronSchedule) > 0
 	numClusters := (int16)(len(domainEntry.GetReplicationConfig().Clusters))
+	updateTimestamp := executionInfo.LastUpdatedTimestamp
 
 	startEvent, err := mutableState.GetStartEvent(ctx)
 	if err != nil {
@@ -473,6 +474,7 @@ func (t *transferActiveTaskExecutor) processCloseExecutionTaskHelper(
 			executionInfo.TaskList,
 			isCron,
 			numClusters,
+			updateTimestamp.UnixNano(),
 			searchAttr,
 		); err != nil {
 			return err
@@ -972,6 +974,7 @@ func (t *transferActiveTaskExecutor) processRecordWorkflowStartedOrUpsertHelper(
 	searchAttr := copySearchAttributes(executionInfo.SearchAttributes)
 	isCron := len(executionInfo.CronSchedule) > 0
 	numClusters := (int16)(len(domainEntry.GetReplicationConfig().Clusters))
+	updateTimestamp := getWorkflowLastUpdatedTimestamp(mutableState)
 
 	// release the context lock since we no longer need mutable state builder and
 	// the rest of logic is making RPC call, which takes time.
@@ -1009,6 +1012,7 @@ func (t *transferActiveTaskExecutor) processRecordWorkflowStartedOrUpsertHelper(
 		visibilityMemo,
 		isCron,
 		numClusters,
+		updateTimestamp.UnixNano(),
 		searchAttr,
 	)
 }

service/history/task/transfer_active_task_executor_test.go

@@ -1823,6 +1823,7 @@ func createUpsertWorkflowSearchAttributesRequest(
 		TaskList:           taskInfo.TaskList,
 		IsCron:             len(executionInfo.CronSchedule) > 0,
 		NumClusters:        numClusters,
+		UpdateTimestamp:    executionInfo.LastUpdatedTimestamp.UnixNano(),
 	}
 }
 

service/history/task/transfer_standby_task_executor.go

@@ -248,6 +248,7 @@ func (t *transferStandbyTaskExecutor) processCloseExecution(
 		visibilityMemo := getWorkflowMemo(executionInfo.Memo)
 		searchAttr := executionInfo.SearchAttributes
 		isCron := len(executionInfo.CronSchedule) > 0
+		updateTimestamp := getWorkflowLastUpdatedTimestamp(mutableState)
 
 		lastWriteVersion, err := mutableState.GetLastWriteVersion()
 		if err != nil {
@@ -282,6 +283,7 @@ func (t *transferStandbyTaskExecutor) processCloseExecution(
 			executionInfo.TaskList,
 			isCron,
 			numClusters,
+			updateTimestamp.UnixNano(),
 			searchAttr,
 		)
 	}
@@ -476,6 +478,7 @@ func (t *transferStandbyTaskExecutor) processRecordWorkflowStartedOrUpsertHelper
 	visibilityMemo := getWorkflowMemo(executionInfo.Memo)
 	searchAttr := copySearchAttributes(executionInfo.SearchAttributes)
 	isCron := len(executionInfo.CronSchedule) > 0
+	updateTimestamp := getWorkflowLastUpdatedTimestamp(mutableState)
 
 	domainEntry, err := t.shard.GetDomainCache().GetDomainByID(transferTask.DomainID)
 	if err != nil {
@@ -515,6 +518,7 @@ func (t *transferStandbyTaskExecutor) processRecordWorkflowStartedOrUpsertHelper
 		visibilityMemo,
 		isCron,
 		numClusters,
+		updateTimestamp.UnixNano(),
 		searchAttr,
 	)
 

service/history/task/transfer_task_executor_base.go

@@ -220,6 +220,7 @@ func (t *transferTaskExecutorBase) upsertWorkflowExecution(
 	visibilityMemo *types.Memo,
 	isCron bool,
 	numClusters int16,
+	updateTimeUnixNano int64,
 	searchAttributes map[string][]byte,
 ) error {
 
@@ -248,6 +249,7 @@ func (t *transferTaskExecutorBase) upsertWorkflowExecution(
 		IsCron:             isCron,
 		NumClusters:        numClusters,
 		SearchAttributes:   searchAttributes,
+		UpdateTimestamp:    updateTimeUnixNano,
 	}
 
 	return t.visibilityMgr.UpsertWorkflowExecution(ctx, request)
@@ -269,6 +271,7 @@ func (t *transferTaskExecutorBase) recordWorkflowClosed(
 	taskList string,
 	isCron bool,
 	numClusters int16,
+	updateTimeUnixNano int64,
 	searchAttributes map[string][]byte,
 ) error {
 
@@ -318,6 +321,7 @@ func (t *transferTaskExecutorBase) recordWorkflowClosed(
 			TaskList:           taskList,
 			SearchAttributes:   searchAttributes,
 			IsCron:             isCron,
+			UpdateTimestamp:    updateTimeUnixNano,
 			NumClusters:        numClusters,
 		}); err != nil {
 			return err
@@ -373,6 +377,12 @@ func getWorkflowExecutionTimestamp(
 	return executionTimestamp
 }
 
+func getWorkflowLastUpdatedTimestamp(
+	msBuilder execution.MutableState,
+) time.Time {
+	return msBuilder.GetExecutionInfo().LastUpdatedTimestamp
+}
+
 func getWorkflowMemo(
 	memo map[string][]byte,
 ) *types.Memo {