Implement Admin Get Raw History API For NDC (cadence-workflow#2625)

* Implement re-replication API v2. This API is going to use in re-replication case. It takes in start event id, start event version and end event id, end event version exclusive-exclusive.
MirrorFork · Oct 10, 2019 · 189b1e1 · 189b1e1
1 parent 0211497
commit 189b1e1
Show file tree

Hide file tree

Showing 17 changed files with 3,222 additions and 597 deletions.
diff --git a/.gen/go/admin/admin.go b/.gen/go/admin/admin.go
diff --git a/.gen/go/admin/adminserviceclient/client.go b/.gen/go/admin/adminserviceclient/client.go
diff --git a/.gen/go/admin/adminserviceserver/server.go b/.gen/go/admin/adminserviceserver/server.go
diff --git a/.gen/go/admin/adminservicetest/client.go b/.gen/go/admin/adminservicetest/client.go
diff --git a/client/admin/client.go b/client/admin/client.go
@@ -151,6 +151,22 @@ func (c *clientImpl) GetWorkflowExecutionRawHistory(
 	return client.GetWorkflowExecutionRawHistory(ctx, request, opts...)
 }
 
+func (c *clientImpl) GetWorkflowExecutionRawHistoryV2(
+	ctx context.Context,
+	request *admin.GetWorkflowExecutionRawHistoryV2Request,
+	opts ...yarpc.CallOption,
+) (*admin.GetWorkflowExecutionRawHistoryV2Response, error) {
+
+	opts = common.AggregateYarpcOptions(ctx, opts...)
+	client, err := c.getRandomClient()
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := c.createContext(ctx)
+	defer cancel()
+	return client.GetWorkflowExecutionRawHistoryV2(ctx, request, opts...)
+}
+
 func (c *clientImpl) createContext(parent context.Context) (context.Context, context.CancelFunc) {
 	if parent == nil {
 		return context.WithTimeout(context.Background(), c.timeout)

diff --git a/client/admin/metricClient.go b/client/admin/metricClient.go
@@ -150,3 +150,21 @@ func (c *metricClient) GetWorkflowExecutionRawHistory(
 	}
 	return resp, err
 }
+
+func (c *metricClient) GetWorkflowExecutionRawHistoryV2(
+	ctx context.Context,
+	request *admin.GetWorkflowExecutionRawHistoryV2Request,
+	opts ...yarpc.CallOption,
+) (*admin.GetWorkflowExecutionRawHistoryV2Response, error) {
+
+	c.metricsClient.IncCounter(metrics.AdminClientGetWorkflowExecutionRawHistoryV2Scope, metrics.CadenceClientRequests)
+
+	sw := c.metricsClient.StartTimer(metrics.AdminClientGetWorkflowExecutionRawHistoryV2Scope, metrics.CadenceClientLatency)
+	resp, err := c.client.GetWorkflowExecutionRawHistoryV2(ctx, request, opts...)
+	sw.Stop()
+
+	if err != nil {
+		c.metricsClient.IncCounter(metrics.AdminClientGetWorkflowExecutionRawHistoryV2Scope, metrics.CadenceClientFailures)
+	}
+	return resp, err
+}
diff --git a/client/admin/retryableClient.go b/client/admin/retryableClient.go
@@ -135,3 +135,19 @@ func (c *retryableClient) GetWorkflowExecutionRawHistory(
 	err := backoff.Retry(op, c.policy, c.isRetryable)
 	return resp, err
 }
+
+func (c *retryableClient) GetWorkflowExecutionRawHistoryV2(
+	ctx context.Context,
+	request *admin.GetWorkflowExecutionRawHistoryV2Request,
+	opts ...yarpc.CallOption,
+) (*admin.GetWorkflowExecutionRawHistoryV2Response, error) {
+
+	var resp *admin.GetWorkflowExecutionRawHistoryV2Response
+	op := func() error {
+		var err error
+		resp, err = c.client.GetWorkflowExecutionRawHistoryV2(ctx, request, opts...)
+		return err
+	}
+	err := backoff.Retry(op, c.policy, c.isRetryable)
+	return resp, err
+}
diff --git a/common/metrics/defs.go b/common/metrics/defs.go
@@ -405,6 +405,8 @@ const (
 	AdminClientDescribeWorkflowExecutionScope
 	// AdminClientGetWorkflowExecutionRawHistoryScope tracks RPC calls to admin service
 	AdminClientGetWorkflowExecutionRawHistoryScope
+	// AdminClientGetWorkflowExecutionRawHistoryV2Scope tracks RPC calls to admin service
+	AdminClientGetWorkflowExecutionRawHistoryV2Scope
 	// DCRedirectionDeprecateDomainScope tracks RPC calls for dc redirection
 	DCRedirectionDeprecateDomainScope
 	// DCRedirectionDescribeDomainScope tracks RPC calls for dc redirection
@@ -580,6 +582,8 @@ const (
 	AdminDescribeWorkflowExecutionScope
 	// AdminGetWorkflowExecutionRawHistoryScope is the metric scope for admin.GetWorkflowExecutionRawHistoryScope
 	AdminGetWorkflowExecutionRawHistoryScope
+	// AdminGetWorkflowExecutionRawHistoryV2Scope is the metric scope for admin.GetWorkflowExecutionRawHistoryScope
+	AdminGetWorkflowExecutionRawHistoryV2Scope
 	// AdminRemoveTaskScope is the metric scope for admin.AdminRemoveTaskScope
 	AdminRemoveTaskScope
 	//AdminCloseShardTaskScope is the metric scope for admin.AdminRemoveTaskScope
@@ -1080,6 +1084,7 @@ var ScopeDefs = map[ServiceIdx]map[int]scopeDefinition{
 		AdminClientDescribeHistoryHostScope:                 {operation: "AdminClientDescribeHistoryHost", tags: map[string]string{CadenceRoleTagName: AdminRoleTagValue}},
 		AdminClientDescribeWorkflowExecutionScope:           {operation: "AdminClientDescribeWorkflowExecution", tags: map[string]string{CadenceRoleTagName: AdminRoleTagValue}},
 		AdminClientGetWorkflowExecutionRawHistoryScope:      {operation: "AdminClientGetWorkflowExecutionRawHistory", tags: map[string]string{CadenceRoleTagName: AdminRoleTagValue}},
+		AdminClientGetWorkflowExecutionRawHistoryV2Scope:    {operation: "AdminClientGetWorkflowExecutionRawHistoryV2", tags: map[string]string{CadenceRoleTagName: AdminRoleTagValue}},
 		AdminClientCloseShardScope:                          {operation: "AdminClientCloseShard", tags: map[string]string{CadenceRoleTagName: AdminRoleTagValue}},
 		DCRedirectionDeprecateDomainScope:                   {operation: "DCRedirectionDeprecateDomain", tags: map[string]string{CadenceRoleTagName: DCRedirectionRoleTagValue}},
 		DCRedirectionDescribeDomainScope:                    {operation: "DCRedirectionDescribeDomain", tags: map[string]string{CadenceRoleTagName: DCRedirectionRoleTagValue}},
@@ -1158,11 +1163,12 @@ var ScopeDefs = map[ServiceIdx]map[int]scopeDefinition{
 	// Frontend Scope Names
 	Frontend: {
 		// Admin API scope co-locates with with frontend
-		AdminRemoveTaskScope:                     {operation: "AdminRemoveTask"},
-		AdminCloseShardTaskScope:                 {operation: "AdminCloseShardTask"},
-		AdminDescribeHistoryHostScope:            {operation: "DescribeHistoryHost"},
-		AdminDescribeWorkflowExecutionScope:      {operation: "DescribeWorkflowExecution"},
-		AdminGetWorkflowExecutionRawHistoryScope: {operation: "GetWorkflowExecutionRawHistory"},
+		AdminRemoveTaskScope:                       {operation: "AdminRemoveTask"},
+		AdminCloseShardTaskScope:                   {operation: "AdminCloseShardTask"},
+		AdminDescribeHistoryHostScope:              {operation: "DescribeHistoryHost"},
+		AdminDescribeWorkflowExecutionScope:        {operation: "DescribeWorkflowExecution"},
+		AdminGetWorkflowExecutionRawHistoryScope:   {operation: "GetWorkflowExecutionRawHistory"},
+		AdminGetWorkflowExecutionRawHistoryV2Scope: {operation: "GetWorkflowExecutionRawHistoryV2"},
 
 		FrontendStartWorkflowExecutionScope:           {operation: "StartWorkflowExecution"},
 		FrontendPollForDecisionTaskScope:              {operation: "PollForDecisionTask"},

diff --git a/common/persistence/versionHistory.go b/common/persistence/versionHistory.go
@@ -90,7 +90,7 @@ func (item *VersionHistoryItem) Equals(input *VersionHistoryItem) bool {
 // NewVersionHistory create a new version history
 func NewVersionHistory(
 	inputToken []byte,
-	inputitems []*VersionHistoryItem,
+	inputItems []*VersionHistoryItem,
 ) *VersionHistory {
 
 	token := make([]byte, len(inputToken))
@@ -100,7 +100,7 @@ func NewVersionHistory(
 		items:       nil,
 	}
 
-	for _, item := range inputitems {
+	for _, item := range inputItems {
 		if err := versionHistory.AddOrUpdateItem(item.Duplicate()); err != nil {
 			panic(fmt.Sprintf("unable to initialize version history: %v", err))
 		}
@@ -243,6 +243,10 @@ func (v *VersionHistory) ContainsItem(
 	prevEventID := common.FirstEventID - 1
 	for _, currentItem := range v.items {
 		if item.GetVersion() == currentItem.GetVersion() {
+			// this is a special handling for event id = 0
+			if (item.GetEventID() == common.FirstEventID-1) && item.GetEventID() <= currentItem.GetEventID() {
+				return true
+			}
 			if prevEventID < item.GetEventID() && item.GetEventID() <= currentItem.GetEventID() {
 				return true
 			}

diff --git a/common/persistence/versionHistory_test.go b/common/persistence/versionHistory_test.go
@@ -280,6 +280,17 @@ func (s *versionHistoriesSuite) TestContainsItem_False() {
 	s.False(history.ContainsItem(NewVersionHistoryItem(6, 5)))
 }
 
+func (s *versionHistoriesSuite) TestContainsItem_True_WhenFirstEventIDIsZero() {
+	branchToken := []byte("some random branch token")
+	items := []*VersionHistoryItem{
+		{eventID: 3, version: 0},
+		{eventID: 6, version: 4},
+	}
+	history := NewVersionHistory(branchToken, items)
+
+	s.True(history.ContainsItem(NewVersionHistoryItem(common.FirstEventID-1, 0)))
+}
+
 func (s *versionHistorySuite) TestIsLCAAppendable_True() {
 	branchToken := []byte("some random branch token")
 	items := []*VersionHistoryItem{

diff --git a/common/xdc/historyRereplicator.go b/common/xdc/historyRereplicator.go
@@ -384,8 +384,15 @@ func (c *historyRereplicationContext) handleEmptyHistory(domainID string, workfl
 	return err
 }
 
-func (c *historyRereplicationContext) getHistory(domainID string, workflowID string, runID string,
-	firstEventID int64, nextEventID int64, token []byte, pageSize int32) (*admin.GetWorkflowExecutionRawHistoryResponse, error) {
+func (c *historyRereplicationContext) getHistory(
+	domainID string,
+	workflowID string,
+	runID string,
+	firstEventID int64,
+	nextEventID int64,
+	token []byte,
+	pageSize int32,
+) (*admin.GetWorkflowExecutionRawHistoryResponse, error) {
 
 	logger := c.logger.WithTags(tag.WorkflowRunID(runID), tag.WorkflowFirstEventID(firstEventID), tag.WorkflowNextEventID(nextEventID))