Adding auto-reset points to visibility record (cadence-workflow#2830)

* Adding auto-reset points to visibility record

* avoid initializing map in the constructor

common/definition/indexedKeys.go

@@ -24,17 +24,18 @@ import "github.com/uber/cadence/.gen/go/shared"
 
 // valid indexed fields on ES
 const (
-	DomainID      = "DomainID"
-	WorkflowID    = "WorkflowID"
-	RunID         = "RunID"
-	WorkflowType  = "WorkflowType"
-	StartTime     = "StartTime"
-	ExecutionTime = "ExecutionTime"
-	CloseTime     = "CloseTime"
-	CloseStatus   = "CloseStatus"
-	HistoryLength = "HistoryLength"
-	Encoding      = "Encoding"
-	KafkaKey      = "KafkaKey"
+	DomainID        = "DomainID"
+	WorkflowID      = "WorkflowID"
+	RunID           = "RunID"
+	WorkflowType    = "WorkflowType"
+	StartTime       = "StartTime"
+	ExecutionTime   = "ExecutionTime"
+	CloseTime       = "CloseTime"
+	CloseStatus     = "CloseStatus"
+	HistoryLength   = "HistoryLength"
+	Encoding        = "Encoding"
+	KafkaKey        = "KafkaKey"
+	BinaryChecksums = "BinaryChecksums"
 
 	CustomStringField    = "CustomStringField"
 	CustomKeywordField   = "CustomKeywordField"
@@ -65,6 +66,7 @@ func createDefaultIndexedKeys() map[string]interface{} {
 		CustomDoubleField:    shared.IndexedValueTypeDouble,
 		CustomDatetimeField:  shared.IndexedValueTypeDatetime,
 		CadenceChangeVersion: shared.IndexedValueTypeKeyword,
+		BinaryChecksums:      shared.IndexedValueTypeKeyword,
 	}
 	for k, v := range systemIndexedKeys {
 		defaultIndexedKeys[k] = v

config/dynamicconfig/development_es.yaml

@@ -31,5 +31,6 @@ frontend.validSearchAttributes:
       Operator: 1
       RolloutID: 1
       CadenceChangeVersion: 1
+      BinaryChecksums: 1
 system.minRetentionDays:
     - value: 0

host/elasticsearch_test.go

@@ -226,7 +226,7 @@ func (s *elasticsearchIntegrationSuite) TestListWorkflow_SearchAttribute() {
 	listRequest := &workflow.ListWorkflowExecutionsRequest{
 		Domain:   common.StringPtr(s.domainName),
 		PageSize: common.Int32Ptr(int32(2)),
-		Query:    common.StringPtr(fmt.Sprintf(`WorkflowType = '%s' and CloseTime = missing`, wt)),
+		Query:    common.StringPtr(fmt.Sprintf(`WorkflowType = '%s' and CloseTime = missing and BinaryChecksums = 'binary-v1'`, wt)),
 	}
 	// verify upsert data is on ES
 	s.testListResultForUpsertSearchAttributes(listRequest)
@@ -970,18 +970,26 @@ func (s *elasticsearchIntegrationSuite) testListResultForUpsertSearchAttributes(
 		if len(resp.GetExecutions()) == 1 {
 			execution := resp.GetExecutions()[0]
 			retrievedSearchAttr := execution.SearchAttributes
-			if retrievedSearchAttr != nil && len(retrievedSearchAttr.GetIndexedFields()) == 2 {
+			if retrievedSearchAttr != nil && len(retrievedSearchAttr.GetIndexedFields()) == 3 {
 				fields := retrievedSearchAttr.GetIndexedFields()
 				searchValBytes := fields[s.testSearchAttributeKey]
 				var searchVal string
-				json.Unmarshal(searchValBytes, &searchVal)
+				err := json.Unmarshal(searchValBytes, &searchVal)
+				s.Nil(err)
 				s.Equal("another string", searchVal)
 
 				searchValBytes2 := fields[definition.CustomIntField]
 				var searchVal2 int
-				json.Unmarshal(searchValBytes2, &searchVal2)
+				err = json.Unmarshal(searchValBytes2, &searchVal2)
+				s.Nil(err)
 				s.Equal(123, searchVal2)
 
+				binaryChecksumsBytes := fields[definition.BinaryChecksums]
+				var binaryChecksums []string
+				err = json.Unmarshal(binaryChecksumsBytes, &binaryChecksums)
+				s.Nil(err)
+				s.Equal([]string{"binary-v1", "binary-v2"}, binaryChecksums)
+
 				verified = true
 				break
 			}
@@ -994,10 +1002,12 @@ func (s *elasticsearchIntegrationSuite) testListResultForUpsertSearchAttributes(
 func getUpsertSearchAttributes() *workflow.SearchAttributes {
 	attrValBytes1, _ := json.Marshal("another string")
 	attrValBytes2, _ := json.Marshal(123)
+	binaryChecksums, _ := json.Marshal([]string{"binary-v1", "binary-v2"})
 	upsertSearchAttr := &workflow.SearchAttributes{
 		IndexedFields: map[string][]byte{
 			definition.CustomStringField: attrValBytes1,
 			definition.CustomIntField:    attrValBytes2,
+			definition.BinaryChecksums:   binaryChecksums,
 		},
 	}
 	return upsertSearchAttr

host/testdata/es_index_template.json

@@ -45,12 +45,23 @@
         },
         "Attr": {
           "properties": {
+            "CadenceChangeVersion":  { "type": "keyword" },
             "CustomStringField":  { "type": "text" },
             "CustomKeywordField": { "type": "keyword"},
             "CustomIntField": { "type": "long"},
             "CustomBoolField": { "type": "boolean"},
             "CustomDoubleField": { "type": "double"},
-            "CustomDatetimeField": { "type": "date"}
+            "CustomDatetimeField": { "type": "date"},
+            "project": { "type": "keyword"},
+            "service": { "type": "keyword"},
+            "environment": { "type": "keyword"},
+            "addon": { "type": "keyword"},
+            "addon-type": { "type": "keyword"},
+            "user": { "type": "keyword"},
+            "CustomDomain": { "type": "keyword"},
+            "Operator": { "type": "keyword"},
+            "RolloutID": { "type": "keyword"},
+            "BinaryChecksums": { "type": "keyword"}
           }
         }
       }

schema/elasticsearch/visibility/index_template.json

@@ -60,7 +60,8 @@
             "user": { "type": "keyword"},
             "CustomDomain": { "type": "keyword"},
             "Operator": { "type": "keyword"},
-            "RolloutID": { "type": "keyword"}
+            "RolloutID": { "type": "keyword"},
+            "BinaryChecksums": { "type": "keyword"}
           }
         }
       }

service/history/mutableStateBuilder.go

@@ -21,6 +21,7 @@
 package history
 
 import (
+	"encoding/json"
 	"fmt"
 	"time"
 
@@ -33,6 +34,7 @@ import (
 	"github.com/uber/cadence/common/cache"
 	"github.com/uber/cadence/common/clock"
 	"github.com/uber/cadence/common/cluster"
+	"github.com/uber/cadence/common/definition"
 	"github.com/uber/cadence/common/errors"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
@@ -1880,10 +1882,10 @@ func (e *mutableStateBuilder) CreateTransientDecisionEvents(
 func (e *mutableStateBuilder) addBinaryCheckSumIfNotExists(
 	event *workflow.HistoryEvent,
 	maxResetPoints int,
-) {
+) error {
 	binChecksum := event.GetDecisionTaskCompletedEventAttributes().GetBinaryChecksum()
 	if len(binChecksum) == 0 {
-		return
+		return nil
 	}
 	exeInfo := e.executionInfo
 	var currResetPoints []*workflow.ResetPointInfo
@@ -1893,17 +1895,24 @@ func (e *mutableStateBuilder) addBinaryCheckSumIfNotExists(
 		currResetPoints = make([]*workflow.ResetPointInfo, 0, 1)
 	}
 
+	// List of all recent binary checksums associated with the workflow.
+	var recentBinaryChecksums []string
+
 	for _, rp := range currResetPoints {
+		recentBinaryChecksums = append(recentBinaryChecksums, rp.GetBinaryChecksum())
 		if rp.GetBinaryChecksum() == binChecksum {
 			// this checksum already exists
-			return
+			return nil
 		}
 	}
 
 	if len(currResetPoints) == maxResetPoints {
-		// if exceeding the max limit, do rotation by taking the oldest one out
+		// If exceeding the max limit, do rotation by taking the oldest one out.
 		currResetPoints = currResetPoints[1:]
+		recentBinaryChecksums = recentBinaryChecksums[1:]
 	}
+	// Adding current version of the binary checksum.
+	recentBinaryChecksums = append(recentBinaryChecksums, binChecksum)
 
 	resettable := true
 	err := e.CheckResettable()
@@ -1921,6 +1930,18 @@ func (e *mutableStateBuilder) addBinaryCheckSumIfNotExists(
 	exeInfo.AutoResetPoints = &workflow.ResetPoints{
 		Points: currResetPoints,
 	}
+	bytes, err := json.Marshal(recentBinaryChecksums)
+	if err != nil {
+		return err
+	}
+	if exeInfo.SearchAttributes == nil {
+		exeInfo.SearchAttributes = make(map[string][]byte)
+	}
+	exeInfo.SearchAttributes[definition.BinaryChecksums] = bytes
+	if e.shard.GetConfig().AdvancedVisibilityWritingMode() != common.AdvancedVisibilityWritingModeOff {
+		return e.taskGenerator.generateWorkflowSearchAttrTasks(e.unixNanoToTime(event.GetTimestamp()))
+	}
+	return nil
 }
 
 // TODO: we will release the restriction when reset API allow those pending

service/history/mutableStateDecisionTaskManager.go

@@ -235,8 +235,7 @@ func (m *mutableStateDecisionTaskManagerImpl) ReplicateDecisionTaskCompletedEven
 	event *workflow.HistoryEvent,
 ) error {
 	m.beforeAddDecisionTaskCompletedEvent()
-	m.afterAddDecisionTaskCompletedEvent(event, math.MaxInt32)
-	return nil
+	return m.afterAddDecisionTaskCompletedEvent(event, math.MaxInt32)
 }
 
 func (m *mutableStateDecisionTaskManagerImpl) ReplicateDecisionTaskFailedEvent() error {
@@ -477,7 +476,10 @@ func (m *mutableStateDecisionTaskManagerImpl) AddDecisionTaskCompletedEvent(
 	// Now write the completed event
 	event := m.msb.hBuilder.AddDecisionTaskCompletedEvent(scheduleEventID, startedEventID, request)
 
-	m.afterAddDecisionTaskCompletedEvent(event, maxResetPoints)
+	err := m.afterAddDecisionTaskCompletedEvent(event, maxResetPoints)
+	if err != nil {
+		return nil, err
+	}
 	return event, nil
 }
 
@@ -721,7 +723,7 @@ func (m *mutableStateDecisionTaskManagerImpl) beforeAddDecisionTaskCompletedEven
 func (m *mutableStateDecisionTaskManagerImpl) afterAddDecisionTaskCompletedEvent(
 	event *workflow.HistoryEvent,
 	maxResetPoints int,
-) {
+) error {
 	m.msb.executionInfo.LastProcessedEvent = event.GetDecisionTaskCompletedEventAttributes().GetStartedEventId()
-	m.msb.addBinaryCheckSumIfNotExists(event, maxResetPoints)
+	return m.msb.addBinaryCheckSumIfNotExists(event, maxResetPoints)
 }