Feature/archival poc

cmd/server/cadence/server.go

@@ -242,7 +242,11 @@ func (s *server) startService() common.Daemon {
 		&s.cfg.DomainDefaults.Archival,
 	)
 
-	params.ArchiverProvider = provider.NewArchiverProvider(s.cfg.Archival.History.Provider, s.cfg.Archival.Visibility.Provider)
+	params.ArchiverProvider = provider.NewArchiverProvider(
+		s.cfg.Archival.History.Provider,
+		s.cfg.Archival.Visibility.Provider,
+		s.cfg.Archival.Execution.Provider,
+	)
 	params.PersistenceConfig.TransactionSizeLimit = dc.GetIntProperty(dynamicconfig.TransactionSizeLimit)
 	params.PersistenceConfig.ErrorInjectionRate = dc.GetFloat64Property(dynamicconfig.PersistenceErrorInjectionRate)
 	params.AuthorizationConfig = s.cfg.Authorization

common/archiver/constants.go

@@ -60,3 +60,16 @@ var (
 	// ErrHistoryNotExist is the error for non-exist history
 	ErrHistoryNotExist = errors.New("requested workflow history does not exist")
 )
+
+var _ error = ErrInvalidArchivalRequest{}
+
+// ErrInvalidArchivalRequest A generic nonretriable Bad request error for achival
+type ErrInvalidArchivalRequest struct {
+	Message string
+	URI     URI
+	Request interface{}
+}
+
+func (e ErrInvalidArchivalRequest) Error() string {
+	return e.Message
+}

common/archiver/filestore/archiver.go

@@ -0,0 +1,28 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-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
+
+type archiverImpl struct {
+	history   historyArchiver
+	execution executionArchiver
+}

common/archiver/filestore/executionArchiver.go

@@ -0,0 +1,158 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-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"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+
+	"github.com/uber/cadence/common/archiver"
+	"github.com/uber/cadence/common/config"
+	"github.com/uber/cadence/common/log/tag"
+	"github.com/uber/cadence/common/persistence"
+	"github.com/uber/cadence/common/types"
+	"github.com/uber/cadence/common/util"
+)
+
+type executionArchiver struct {
+	container archiver.ExecutionArchiverBootstrapContainer
+
+	fileMode os.FileMode
+	dirMode  os.FileMode
+}
+
+// NewHistoryArchiver creates a new archiver.HistoryArchiver based on filestore
+func NewExecutionArchiver(
+	container archiver.ExecutionArchiverBootstrapContainer,
+	config *config.FilestoreArchiver,
+) (archiver.ExecutionArchiver, error) {
+	return newExecutionArchiver(container, config)
+}
+
+func newExecutionArchiver(
+	container archiver.ExecutionArchiverBootstrapContainer,
+	config *config.FilestoreArchiver,
+) (*executionArchiver, error) {
+	fileMode, err := strconv.ParseUint(config.FileMode, 0, 32)
+	if err != nil {
+		return nil, errInvalidFileMode
+	}
+	dirMode, err := strconv.ParseUint(config.DirMode, 0, 32)
+	if err != nil {
+		return nil, errInvalidDirMode
+	}
+	return &executionArchiver{
+		container: container,
+		fileMode:  os.FileMode(fileMode),
+		dirMode:   os.FileMode(dirMode),
+	}, nil
+}
+
+func (e *executionArchiver) ValidateExecutionURI(URI archiver.URI) error {
+	if URI == nil || URI.Path() == "" {
+		return archiver.ErrInvalidURI
+	}
+	return nil
+}
+
+func (e *executionArchiver) ValidateExecutionArchiveRequest(req *archiver.ArchiveHistoryRequest, uri archiver.URI) error {
+	if req == nil {
+		return &archiver.ErrInvalidArchivalRequest{
+			Message: "Invalid execution archival request",
+			URI:     uri,
+			Request: req,
+		}
+	}
+	return nil
+}
+
+func (e *executionArchiver) ArchiveExecution(ctx context.Context, URI archiver.URI, req *archiver.ArchiveExecutionRequest, opts ...archiver.ArchiveOption) (err error) {
+
+	featureCatalog := archiver.GetFeatureCatalog(opts...)
+	defer func() {
+		if err != nil && !persistence.IsTransientError(err) && featureCatalog.NonRetriableError != nil {
+			err = featureCatalog.NonRetriableError()
+		}
+	}()
+
+	logger := archiver.TagLoggerWithArchiveExecutionRequestAndURI(e.container.Logger, req, URI.String())
+
+	if err := e.ValidateExecutionURI(URI); err != nil {
+		logger.Error(archiver.ArchiveNonRetriableErrorMsg, tag.ArchivalArchiveFailReason(archiver.ErrReasonInvalidURI), tag.Error(err))
+		return err
+	}
+
+	if err := archiver.ValidateExecutionArchiveRequest(req); err != nil {
+		logger.Error(archiver.ArchiveNonRetriableErrorMsg, tag.ArchivalArchiveFailReason(archiver.ErrReasonInvalidArchiveRequest), tag.Error(err))
+		return err
+	}
+
+	executionMgr, err := e.container.ExecutionMgr(req.ShardID)
+	if err != nil {
+		return fmt.Errorf("could not get shard manager: %w", err)
+	}
+
+	execution, err := executionMgr.GetWorkflowExecution(ctx, &persistence.GetWorkflowExecutionRequest{
+		DomainID: req.DomainID,
+		Execution: types.WorkflowExecution{
+			WorkflowID: req.WorkflowID,
+			RunID:      req.RunID,
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to get workflow for archival: %w", err)
+	}
+
+	err = archiver.ValidateExecutionArchiveResponse(execution)
+	if err != nil {
+		logger.Error("couldn't not archive workflow, it failed validation", tag.Error(err))
+		return fmt.Errorf("could not archive workflow execution: %w", err)
+	}
+
+	jsonEncoded, err := json.Marshal(execution.State)
+	if err != nil {
+		logger.Error("couldn't not archive workflow, encoding failed", tag.Error(err))
+		return fmt.Errorf("could not encode workflow execution for archival: %w", err)
+	}
+
+	dirPath := URI.Path()
+
+	wfPath := constructExecutionFilename(req.DomainID, req.WorkflowID, req.RunID)
+	err = util.MkdirAll(dirPath, e.dirMode)
+	if err != nil {
+		logger.Error("couldn't not create folder for archival", tag.Error(err), tag.Dynamic("filepath", dirPath))
+		return fmt.Errorf("could not create folder for archival: %w", err)
+	}
+
+	err = util.WriteFile(filepath.Join(dirPath, wfPath), jsonEncoded, e.fileMode)
+	if err != nil {
+		logger.Error("failed to write file for archival", tag.Error(err))
+		return fmt.Errorf("failed to write file to %q, error: %w", filepath.Join(dirPath, wfPath), err)
+	}
+
+	return nil
+}

common/archiver/filestore/executionReader.go

@@ -0,0 +1,61 @@
+package filestore
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/uber/cadence/common/archiver"
+	"github.com/uber/cadence/common/cache"
+	"github.com/uber/cadence/common/log"
+	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/persistence"
+	"github.com/uber/cadence/common/util"
+	"path/filepath"
+)
+
+type executionReader struct {
+	log         log.Logger
+	metrics     metrics.Client
+	domainCache cache.DomainCache
+}
+
+// NewExecutionReader is used for reading execution values
+func NewExecutionReader(
+	logger log.Logger,
+	metrics metrics.Client,
+	domainCache cache.DomainCache,
+) (archiver.ExecutionReader, error) {
+	return &executionReader{
+		log:         logger,
+		metrics:     metrics,
+		domainCache: domainCache,
+	}, nil
+}
+
+func (e *executionReader) GetWorkflowExecutionForPersistence(ctx context.Context, req *archiver.GetExecutionRequest) (*archiver.GetExecutionResponse, error) {
+	domain, err := e.domainCache.GetDomainByID(req.DomainID)
+	if err != nil {
+		return nil, err
+	}
+
+	uri, err := archiver.NewURI(domain.GetConfig().HistoryArchivalURI)
+	if err != nil {
+		return nil, err
+	}
+	return e.GetWorkflowExecution(ctx, uri, req)
+}
+
+func (e *executionReader) GetWorkflowExecution(ctx context.Context, URI archiver.URI, req *archiver.GetExecutionRequest) (*archiver.GetExecutionResponse, error) {
+	dirPath := URI.Path()
+	wfPath := constructExecutionFilename(req.DomainID, req.WorkflowID, req.RunID)
+	data, err := util.ReadFile(filepath.Join(dirPath, wfPath))
+	if err != nil {
+		return nil, fmt.Errorf("failed to read workflow execution from archived filepath: %w", err)
+	}
+
+	var ms persistence.WorkflowMutableState
+	json.Unmarshal(data, &ms)
+	return &archiver.GetExecutionResponse{
+		MutableState: &ms,
+	}, nil
+}

common/archiver/filestore/historyArchiver.go

@@ -193,6 +193,21 @@ func (h *historyArchiver) Archive(
 	return nil
 }
 
+func (h *historyArchiver) GetWorkflowHistoryForPersistence(
+	ctx context.Context,
+	request *archiver.GetHistoryRequest,
+) (*archiver.GetHistoryResponse, error) {
+	domain, err := h.container.DomainCache.GetDomainByID(request.DomainID)
+	if err != nil {
+		return nil, err
+	}
+	uri, err := archiver.NewURI(domain.GetConfig().HistoryArchivalURI)
+	if err != nil {
+		return nil, err
+	}
+	return h.Get(ctx, uri, request)
+}
+
 func (h *historyArchiver) Get(
 	ctx context.Context,
 	URI archiver.URI,
@@ -313,7 +328,7 @@ func getNextHistoryBlob(ctx context.Context, historyIterator archiver.HistoryIte
 }
 
 func getHighestVersion(dirPath string, request *archiver.GetHistoryRequest) (*int64, error) {
-	filenames, err := util.ListFilesByPrefix(dirPath, constructHistoryFilenamePrefix(request.DomainID, request.WorkflowID, request.RunID))
+	filenames, err := util.ListFilesByPrefix(dirPath, constructFilenamePrefix(request.DomainID, request.WorkflowID, request.RunID))
 	if err != nil {
 		return nil, err
 	}

common/archiver/filestore/util.go

@@ -85,11 +85,16 @@ func deserializeQueryVisibilityToken(bytes []byte) (*queryVisibilityToken, error
 // File name construction
 
 func constructHistoryFilename(domainID, workflowID, runID string, version int64) string {
-	combinedHash := constructHistoryFilenamePrefix(domainID, workflowID, runID)
+	combinedHash := constructFilenamePrefix(domainID, workflowID, runID)
 	return fmt.Sprintf("%s_%v.history", combinedHash, version)
 }
 
-func constructHistoryFilenamePrefix(domainID, workflowID, runID string) string {
+func constructExecutionFilename(domainID, workflowID, runID string) string {
+	combinedHash := constructFilenamePrefix(domainID, workflowID, runID)
+	return fmt.Sprintf("%s_.concrete_execution", combinedHash)
+}
+
+func constructFilenamePrefix(domainID, workflowID, runID string) string {
 	return strings.Join([]string{hash(domainID), hash(workflowID), hash(runID)}, "")
 }