mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
Michael SnowdenandGitHub
d5c62e926f
<!-- Describe what has changed in this PR --> **What changed?** I renamed all of our Go source code files so that they're snake case instead of camel case. <!-- Tell your future self why have you made these changes --> **Why?** This was an annoying inconsistency we carried over from much older code in the repo. <!-- How have you verified this change? Tested locally? Added a unit test? Checked in staging env? --> **How did you test it?** The renaming process was part of an automated script (https://gist.github.com/MichaelSnowden/47576fffe27aa1cae6b26eaccec7ab55), so it should be consistent. Also, I don't see us doing anything in our codebase that involves parsing our own source files by path: <img width="687" alt="image" src="https://github.com/temporalio/temporal/assets/5942963/c6b873b0-d841-40d1-8b0f-2fe9d8079a53"> *PS: here's a link to the GPT convo I had to generate this renaming script: https://chat.openai.com/share/7f2d011a-a8da-4fa4-9c6d-e09102fdd49f* <!-- Assuming the worst case, what can be broken when deploying this change to production? --> **Potential risks** Mainly merge conflicts, but I used `git mv` so that these are renames instead of deletes/adds, so that shouldn't be a big issue. <!-- Is this PR a hotfix candidate or require that a notification be sent to the broader community? (Yes/No) --> **Is hotfix candidate?** No.
647 lines
20 KiB
Go
647 lines
20 KiB
Go
// The MIT License
|
|
//
|
|
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
|
|
//
|
|
// Copyright (c) 2020 Uber Technologies, Inc.
|
|
//
|
|
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
// of this software and associated documentation files (the "Software"), to deal
|
|
// in the Software without restriction, including without limitation the rights
|
|
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
// copies of the Software, and to permit persons to whom the Software is
|
|
// furnished to do so, subject to the following conditions:
|
|
//
|
|
// The above copyright notice and this permission notice shall be included in
|
|
// all copies or substantial portions of the Software.
|
|
//
|
|
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
// THE SOFTWARE.
|
|
|
|
package filestore
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"path"
|
|
"testing"
|
|
"time"
|
|
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
|
|
"go.temporal.io/server/tests/testutils"
|
|
|
|
"github.com/golang/mock/gomock"
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/stretchr/testify/suite"
|
|
historypb "go.temporal.io/api/history/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
|
|
archiverspb "go.temporal.io/server/api/archiver/v1"
|
|
"go.temporal.io/server/common"
|
|
"go.temporal.io/server/common/archiver"
|
|
"go.temporal.io/server/common/config"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/primitives/timestamp"
|
|
)
|
|
|
|
const (
|
|
testNamespaceID = "test-namespace-id"
|
|
testNamespace = "test-namespace"
|
|
testWorkflowID = "test-workflow-id"
|
|
testRunID = "test-run-id"
|
|
testNextEventID = 1800
|
|
testCloseFailoverVersion = int64(100)
|
|
testPageSize = 100
|
|
|
|
testFileModeStr = "0666"
|
|
testDirModeStr = "0766"
|
|
)
|
|
|
|
var (
|
|
testBranchToken = []byte{1, 2, 3}
|
|
)
|
|
|
|
type historyArchiverSuite struct {
|
|
*require.Assertions
|
|
suite.Suite
|
|
|
|
container *archiver.HistoryBootstrapContainer
|
|
testArchivalURI archiver.URI
|
|
testGetDirectory string
|
|
historyBatchesV1 []*historypb.History
|
|
historyBatchesV100 []*historypb.History
|
|
}
|
|
|
|
func TestHistoryArchiverSuite(t *testing.T) {
|
|
suite.Run(t, new(historyArchiverSuite))
|
|
}
|
|
|
|
func (s *historyArchiverSuite) SetupSuite() {
|
|
var err error
|
|
s.testGetDirectory, err = os.MkdirTemp("", "TestGet")
|
|
s.Require().NoError(err)
|
|
s.setupHistoryDirectory()
|
|
s.testArchivalURI, err = archiver.NewURI("file:///a/b/c")
|
|
s.Require().NoError(err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TearDownSuite() {
|
|
if err := os.RemoveAll(s.testGetDirectory); err != nil {
|
|
s.Fail("Failed to remove test directory %v: %v", s.testGetDirectory, err)
|
|
}
|
|
}
|
|
|
|
func (s *historyArchiverSuite) SetupTest() {
|
|
s.Assertions = require.New(s.T())
|
|
s.container = &archiver.HistoryBootstrapContainer{
|
|
Logger: log.NewNoopLogger(),
|
|
}
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestValidateURI() {
|
|
testCases := []struct {
|
|
URI string
|
|
expectedErr error
|
|
}{
|
|
{
|
|
URI: "wrongscheme:///a/b/c",
|
|
expectedErr: archiver.ErrURISchemeMismatch,
|
|
},
|
|
{
|
|
URI: "file://",
|
|
expectedErr: errEmptyDirectoryPath,
|
|
},
|
|
{
|
|
URI: "file:///a/b/c",
|
|
expectedErr: nil,
|
|
},
|
|
}
|
|
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
for _, tc := range testCases {
|
|
URI, err := archiver.NewURI(tc.URI)
|
|
s.NoError(err)
|
|
s.Equal(tc.expectedErr, historyArchiver.ValidateURI(URI))
|
|
}
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestArchive_Fail_InvalidURI() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
request := &archiver.ArchiveHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
Namespace: testNamespace,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
BranchToken: testBranchToken,
|
|
NextEventID: testNextEventID,
|
|
CloseFailoverVersion: testCloseFailoverVersion,
|
|
}
|
|
URI, err := archiver.NewURI("wrongscheme://")
|
|
s.NoError(err)
|
|
err = historyArchiver.Archive(context.Background(), URI, request)
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestArchive_Fail_InvalidRequest() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
request := &archiver.ArchiveHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
Namespace: testNamespace,
|
|
WorkflowID: "", // an invalid request
|
|
RunID: testRunID,
|
|
BranchToken: testBranchToken,
|
|
NextEventID: testNextEventID,
|
|
CloseFailoverVersion: testCloseFailoverVersion,
|
|
}
|
|
err := historyArchiver.Archive(context.Background(), s.testArchivalURI, request)
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestArchive_Fail_ErrorOnReadHistory() {
|
|
mockCtrl := gomock.NewController(s.T())
|
|
defer mockCtrl.Finish()
|
|
historyIterator := archiver.NewMockHistoryIterator(mockCtrl)
|
|
gomock.InOrder(
|
|
historyIterator.EXPECT().HasNext().Return(true),
|
|
historyIterator.EXPECT().Next(gomock.Any()).Return(nil, errors.New("some random error")),
|
|
)
|
|
|
|
historyArchiver := s.newTestHistoryArchiver(historyIterator)
|
|
request := &archiver.ArchiveHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
Namespace: testNamespace,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
BranchToken: testBranchToken,
|
|
NextEventID: testNextEventID,
|
|
CloseFailoverVersion: testCloseFailoverVersion,
|
|
}
|
|
err := historyArchiver.Archive(context.Background(), s.testArchivalURI, request)
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestArchive_Fail_TimeoutWhenReadingHistory() {
|
|
mockCtrl := gomock.NewController(s.T())
|
|
defer mockCtrl.Finish()
|
|
historyIterator := archiver.NewMockHistoryIterator(mockCtrl)
|
|
gomock.InOrder(
|
|
historyIterator.EXPECT().HasNext().Return(true),
|
|
historyIterator.EXPECT().Next(gomock.Any()).Return(nil, serviceerror.NewResourceExhausted(enumspb.RESOURCE_EXHAUSTED_CAUSE_RPS_LIMIT, "")),
|
|
)
|
|
|
|
historyArchiver := s.newTestHistoryArchiver(historyIterator)
|
|
request := &archiver.ArchiveHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
Namespace: testNamespace,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
BranchToken: testBranchToken,
|
|
NextEventID: testNextEventID,
|
|
CloseFailoverVersion: testCloseFailoverVersion,
|
|
}
|
|
err := historyArchiver.Archive(getCanceledContext(), s.testArchivalURI, request)
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestArchive_Fail_HistoryMutated() {
|
|
mockCtrl := gomock.NewController(s.T())
|
|
defer mockCtrl.Finish()
|
|
historyIterator := archiver.NewMockHistoryIterator(mockCtrl)
|
|
historyBatches := []*historypb.History{
|
|
{
|
|
Events: []*historypb.HistoryEvent{
|
|
{
|
|
EventId: common.FirstEventID + 1,
|
|
EventTime: timestamp.TimePtr(time.Now().UTC()),
|
|
Version: testCloseFailoverVersion + 1,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
historyBlob := &archiverspb.HistoryBlob{
|
|
Header: &archiverspb.HistoryBlobHeader{
|
|
IsLast: true,
|
|
},
|
|
Body: historyBatches,
|
|
}
|
|
gomock.InOrder(
|
|
historyIterator.EXPECT().HasNext().Return(true),
|
|
historyIterator.EXPECT().Next(gomock.Any()).Return(historyBlob, nil),
|
|
)
|
|
|
|
historyArchiver := s.newTestHistoryArchiver(historyIterator)
|
|
request := &archiver.ArchiveHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
Namespace: testNamespace,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
BranchToken: testBranchToken,
|
|
NextEventID: testNextEventID,
|
|
CloseFailoverVersion: testCloseFailoverVersion,
|
|
}
|
|
err := historyArchiver.Archive(context.Background(), s.testArchivalURI, request)
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestArchive_Fail_NonRetryableErrorOption() {
|
|
mockCtrl := gomock.NewController(s.T())
|
|
defer mockCtrl.Finish()
|
|
historyIterator := archiver.NewMockHistoryIterator(mockCtrl)
|
|
gomock.InOrder(
|
|
historyIterator.EXPECT().HasNext().Return(true),
|
|
historyIterator.EXPECT().Next(gomock.Any()).Return(nil, errors.New("some random error")),
|
|
)
|
|
|
|
historyArchiver := s.newTestHistoryArchiver(historyIterator)
|
|
request := &archiver.ArchiveHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
Namespace: testNamespace,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
BranchToken: testBranchToken,
|
|
NextEventID: testNextEventID,
|
|
CloseFailoverVersion: testCloseFailoverVersion,
|
|
}
|
|
nonRetryableErr := errors.New("some non-retryable error")
|
|
err := historyArchiver.Archive(context.Background(), s.testArchivalURI, request, archiver.GetNonRetryableErrorOption(nonRetryableErr))
|
|
s.Equal(nonRetryableErr, err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestArchive_Skip() {
|
|
mockCtrl := gomock.NewController(s.T())
|
|
defer mockCtrl.Finish()
|
|
historyIterator := archiver.NewMockHistoryIterator(mockCtrl)
|
|
historyBlob := &archiverspb.HistoryBlob{
|
|
Header: &archiverspb.HistoryBlobHeader{
|
|
IsLast: false,
|
|
},
|
|
Body: []*historypb.History{
|
|
{
|
|
Events: []*historypb.HistoryEvent{
|
|
{
|
|
EventId: common.FirstEventID,
|
|
EventTime: timestamp.TimePtr(time.Now().UTC()),
|
|
Version: testCloseFailoverVersion,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
gomock.InOrder(
|
|
historyIterator.EXPECT().HasNext().Return(true),
|
|
historyIterator.EXPECT().Next(gomock.Any()).Return(historyBlob, nil),
|
|
historyIterator.EXPECT().HasNext().Return(true),
|
|
historyIterator.EXPECT().Next(gomock.Any()).Return(nil, serviceerror.NewNotFound("workflow not found")),
|
|
)
|
|
|
|
historyArchiver := s.newTestHistoryArchiver(historyIterator)
|
|
request := &archiver.ArchiveHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
Namespace: testNamespace,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
BranchToken: testBranchToken,
|
|
NextEventID: testNextEventID,
|
|
CloseFailoverVersion: testCloseFailoverVersion,
|
|
}
|
|
err := historyArchiver.Archive(context.Background(), s.testArchivalURI, request)
|
|
s.NoError(err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestArchive_Success() {
|
|
mockCtrl := gomock.NewController(s.T())
|
|
defer mockCtrl.Finish()
|
|
historyIterator := archiver.NewMockHistoryIterator(mockCtrl)
|
|
historyBatches := []*historypb.History{
|
|
{
|
|
Events: []*historypb.HistoryEvent{
|
|
{
|
|
EventId: common.FirstEventID + 1,
|
|
EventTime: timestamp.TimePtr(time.Now().UTC()),
|
|
Version: testCloseFailoverVersion,
|
|
},
|
|
{
|
|
EventId: common.FirstEventID + 2,
|
|
EventTime: timestamp.TimePtr(time.Now().UTC()),
|
|
Version: testCloseFailoverVersion,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Events: []*historypb.HistoryEvent{
|
|
{
|
|
EventId: testNextEventID - 1,
|
|
EventTime: timestamp.TimePtr(time.Now().UTC()),
|
|
Version: testCloseFailoverVersion,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
historyBlob := &archiverspb.HistoryBlob{
|
|
Header: &archiverspb.HistoryBlobHeader{
|
|
IsLast: true,
|
|
},
|
|
Body: historyBatches,
|
|
}
|
|
gomock.InOrder(
|
|
historyIterator.EXPECT().HasNext().Return(true),
|
|
historyIterator.EXPECT().Next(gomock.Any()).Return(historyBlob, nil),
|
|
historyIterator.EXPECT().HasNext().Return(false),
|
|
)
|
|
|
|
dir := testutils.MkdirTemp(s.T(), "", "TestArchiveSingleRead")
|
|
|
|
historyArchiver := s.newTestHistoryArchiver(historyIterator)
|
|
request := &archiver.ArchiveHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
Namespace: testNamespace,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
BranchToken: testBranchToken,
|
|
NextEventID: testNextEventID,
|
|
CloseFailoverVersion: testCloseFailoverVersion,
|
|
}
|
|
URI, err := archiver.NewURI("file://" + dir)
|
|
s.NoError(err)
|
|
err = historyArchiver.Archive(context.Background(), URI, request)
|
|
s.NoError(err)
|
|
|
|
expectedFilename := constructHistoryFilename(testNamespaceID, testWorkflowID, testRunID, testCloseFailoverVersion)
|
|
s.assertFileExists(path.Join(dir, expectedFilename))
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestGet_Fail_InvalidURI() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
request := &archiver.GetHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
PageSize: 100,
|
|
}
|
|
URI, err := archiver.NewURI("wrongscheme://")
|
|
s.NoError(err)
|
|
response, err := historyArchiver.Get(context.Background(), URI, request)
|
|
s.Nil(response)
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestGet_Fail_InvalidRequest() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
request := &archiver.GetHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
PageSize: 0, // pageSize should be greater than 0
|
|
}
|
|
response, err := historyArchiver.Get(context.Background(), s.testArchivalURI, request)
|
|
s.Nil(response)
|
|
s.Error(err)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestGet_Fail_DirectoryNotExist() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
request := &archiver.GetHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
PageSize: testPageSize,
|
|
}
|
|
response, err := historyArchiver.Get(context.Background(), s.testArchivalURI, request)
|
|
s.Nil(response)
|
|
s.Error(err)
|
|
s.IsType(&serviceerror.NotFound{}, err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestGet_Fail_InvalidToken() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
request := &archiver.GetHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
PageSize: testPageSize,
|
|
NextPageToken: []byte{'r', 'a', 'n', 'd', 'o', 'm'},
|
|
}
|
|
URI, err := archiver.NewURI("file:///")
|
|
s.NoError(err)
|
|
response, err := historyArchiver.Get(context.Background(), URI, request)
|
|
s.Nil(response)
|
|
s.Error(err)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestGet_Fail_FileNotExist() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
testCloseFailoverVersion := testCloseFailoverVersion
|
|
request := &archiver.GetHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
PageSize: testPageSize,
|
|
CloseFailoverVersion: &testCloseFailoverVersion,
|
|
}
|
|
URI, err := archiver.NewURI("file:///")
|
|
s.NoError(err)
|
|
response, err := historyArchiver.Get(context.Background(), URI, request)
|
|
s.Nil(response)
|
|
s.Error(err)
|
|
s.IsType(&serviceerror.NotFound{}, err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestGet_Success_PickHighestVersion() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
request := &archiver.GetHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
PageSize: testPageSize,
|
|
}
|
|
URI, err := archiver.NewURI("file://" + s.testGetDirectory)
|
|
s.NoError(err)
|
|
response, err := historyArchiver.Get(context.Background(), URI, request)
|
|
s.NoError(err)
|
|
s.Nil(response.NextPageToken)
|
|
s.Equal(s.historyBatchesV100, response.HistoryBatches)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestGet_Success_UseProvidedVersion() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
testCloseFailoverVersion := int64(1)
|
|
request := &archiver.GetHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
PageSize: testPageSize,
|
|
CloseFailoverVersion: &testCloseFailoverVersion,
|
|
}
|
|
URI, err := archiver.NewURI("file://" + s.testGetDirectory)
|
|
s.NoError(err)
|
|
response, err := historyArchiver.Get(context.Background(), URI, request)
|
|
s.NoError(err)
|
|
s.Nil(response.NextPageToken)
|
|
s.Equal(s.historyBatchesV1, response.HistoryBatches)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestGet_Success_SmallPageSize() {
|
|
historyArchiver := s.newTestHistoryArchiver(nil)
|
|
testCloseFailoverVersion := int64(100)
|
|
request := &archiver.GetHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
PageSize: 1,
|
|
CloseFailoverVersion: &testCloseFailoverVersion,
|
|
}
|
|
var combinedHistory []*historypb.History
|
|
|
|
URI, err := archiver.NewURI("file://" + s.testGetDirectory)
|
|
s.NoError(err)
|
|
response, err := historyArchiver.Get(context.Background(), URI, request)
|
|
s.NoError(err)
|
|
s.NotNil(response)
|
|
s.NotNil(response.NextPageToken)
|
|
s.NotNil(response.HistoryBatches)
|
|
s.Len(response.HistoryBatches, 1)
|
|
combinedHistory = append(combinedHistory, response.HistoryBatches...)
|
|
|
|
request.NextPageToken = response.NextPageToken
|
|
response, err = historyArchiver.Get(context.Background(), URI, request)
|
|
s.NoError(err)
|
|
s.NotNil(response)
|
|
s.Nil(response.NextPageToken)
|
|
s.NotNil(response.HistoryBatches)
|
|
s.Len(response.HistoryBatches, 1)
|
|
combinedHistory = append(combinedHistory, response.HistoryBatches...)
|
|
|
|
s.Equal(s.historyBatchesV100, combinedHistory)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) TestArchiveAndGet() {
|
|
mockCtrl := gomock.NewController(s.T())
|
|
defer mockCtrl.Finish()
|
|
historyIterator := archiver.NewMockHistoryIterator(mockCtrl)
|
|
historyBlob := &archiverspb.HistoryBlob{
|
|
Header: &archiverspb.HistoryBlobHeader{
|
|
IsLast: true,
|
|
},
|
|
Body: s.historyBatchesV100,
|
|
}
|
|
gomock.InOrder(
|
|
historyIterator.EXPECT().HasNext().Return(true),
|
|
historyIterator.EXPECT().Next(gomock.Any()).Return(historyBlob, nil),
|
|
historyIterator.EXPECT().HasNext().Return(false),
|
|
)
|
|
|
|
dir := testutils.MkdirTemp(s.T(), "", "TestArchiveAndGet")
|
|
|
|
historyArchiver := s.newTestHistoryArchiver(historyIterator)
|
|
archiveRequest := &archiver.ArchiveHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
Namespace: testNamespace,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
BranchToken: testBranchToken,
|
|
NextEventID: testNextEventID,
|
|
CloseFailoverVersion: testCloseFailoverVersion,
|
|
}
|
|
URI, err := archiver.NewURI("file://" + dir)
|
|
s.NoError(err)
|
|
err = historyArchiver.Archive(context.Background(), URI, archiveRequest)
|
|
s.NoError(err)
|
|
|
|
expectedFilename := constructHistoryFilename(testNamespaceID, testWorkflowID, testRunID, testCloseFailoverVersion)
|
|
s.assertFileExists(path.Join(dir, expectedFilename))
|
|
|
|
getRequest := &archiver.GetHistoryRequest{
|
|
NamespaceID: testNamespaceID,
|
|
WorkflowID: testWorkflowID,
|
|
RunID: testRunID,
|
|
PageSize: testPageSize,
|
|
}
|
|
response, err := historyArchiver.Get(context.Background(), URI, getRequest)
|
|
s.NoError(err)
|
|
s.NotNil(response)
|
|
s.Nil(response.NextPageToken)
|
|
s.Equal(s.historyBatchesV100, response.HistoryBatches)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) newTestHistoryArchiver(historyIterator archiver.HistoryIterator) *historyArchiver {
|
|
config := &config.FilestoreArchiver{
|
|
FileMode: testFileModeStr,
|
|
DirMode: testDirModeStr,
|
|
}
|
|
archiver, err := newHistoryArchiver(s.container, config, historyIterator)
|
|
s.NoError(err)
|
|
return archiver
|
|
}
|
|
|
|
func (s *historyArchiverSuite) setupHistoryDirectory() {
|
|
now := time.Date(2020, 8, 22, 1, 2, 3, 4, time.UTC)
|
|
s.historyBatchesV1 = []*historypb.History{
|
|
{
|
|
Events: []*historypb.HistoryEvent{
|
|
{
|
|
EventId: testNextEventID - 1,
|
|
EventTime: &now,
|
|
Version: 1,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
s.historyBatchesV100 = []*historypb.History{
|
|
{
|
|
Events: []*historypb.HistoryEvent{
|
|
{
|
|
EventId: common.FirstEventID + 1,
|
|
EventTime: &now,
|
|
Version: testCloseFailoverVersion,
|
|
},
|
|
{
|
|
EventId: common.FirstEventID + 1,
|
|
EventTime: &now,
|
|
Version: testCloseFailoverVersion,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Events: []*historypb.HistoryEvent{
|
|
{
|
|
EventId: testNextEventID - 1,
|
|
EventTime: &now,
|
|
Version: testCloseFailoverVersion,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
s.writeHistoryBatchesForGetTest(s.historyBatchesV1, int64(1))
|
|
s.writeHistoryBatchesForGetTest(s.historyBatchesV100, testCloseFailoverVersion)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) writeHistoryBatchesForGetTest(historyBatches []*historypb.History, version int64) {
|
|
data, err := encodeHistories(historyBatches)
|
|
s.Require().NoError(err)
|
|
filename := constructHistoryFilename(testNamespaceID, testWorkflowID, testRunID, version)
|
|
err = writeFile(path.Join(s.testGetDirectory, filename), data, testFileMode)
|
|
s.Require().NoError(err)
|
|
}
|
|
|
|
func (s *historyArchiverSuite) assertFileExists(filepath string) {
|
|
exists, err := fileExists(filepath)
|
|
s.NoError(err)
|
|
s.True(exists)
|
|
}
|
|
|
|
func getCanceledContext() context.Context {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
return ctx
|
|
}
|