Refactor s3 to general blobStore (#576)

Author: longquanzheng

cmd/server/iwf/iwf.go

@@ -24,6 +24,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"github.com/indeedeng/iwf/service/common/blobstore"
 	rawLog "log"
 	"strings"
 	"sync"
@@ -205,16 +206,21 @@ func launchTemporalService(
 	logger log.Logger,
 ) {
 	s3Client := CreateS3Client(config, context.Background())
+	blobStore := blobstore.NewBlobStore(
+		s3Client,
+		config.Interpreter.Temporal.Namespace,
+		config.ExternalStorage,
+		logger,
+	)
 	switch svcName {
 	case serviceAPI:
 		svc := api.NewService(
 			config, unifiedClient, logger.WithTags(tag.Service(svcName)),
-			s3Client,
-			config.Interpreter.Temporal.Namespace+"/",
+			blobStore,
 		)
 		rawLog.Fatal(svc.Run(fmt.Sprintf(":%v", config.Api.Port)))
 	case serviceInterpreter:
-		interpreter := temporal.NewInterpreterWorker(config, temporalClient, isvc.TaskQueue, false, nil, unifiedClient, s3Client)
+		interpreter := temporal.NewInterpreterWorker(config, temporalClient, isvc.TaskQueue, false, nil, unifiedClient, blobStore)
 		interpreter.Start()
 	default:
 		rawLog.Fatalf("Invalid service: %v", svcName)
@@ -231,16 +237,21 @@ func launchCadenceService(
 	logger log.Logger,
 ) {
 	s3Client := CreateS3Client(config, context.Background())
+	blobStore := blobstore.NewBlobStore(
+		s3Client,
+		config.Interpreter.Cadence.Domain,
+		config.ExternalStorage,
+		logger,
+	)
 	switch svcName {
 	case serviceAPI:
 		svc := api.NewService(
 			config, unifiedClient, logger.WithTags(tag.Service(svcName)),
-			s3Client,
-			config.Interpreter.Cadence.Domain+"/",
+			blobStore,
 		)
 		rawLog.Fatal(svc.Run(fmt.Sprintf(":%v", config.Api.Port)))
 	case serviceInterpreter:
-		interpreter := cadence.NewInterpreterWorker(config, service, domain, isvc.TaskQueue, closeFunc, unifiedClient, s3Client)
+		interpreter := cadence.NewInterpreterWorker(config, service, domain, isvc.TaskQueue, closeFunc, unifiedClient, blobStore)
 		interpreter.Start()
 	default:
 		rawLog.Fatalf("Invalid service: %v", svcName)

config/config.go

@@ -18,6 +18,10 @@ const (
 	StorageStatusInactive = "inactive"
 )
 
+const (
+	StorageTypeS3 = "s3"
+)
+
 type (
 	Config struct {
 		// Log is the logging config
@@ -42,6 +46,7 @@ type (
 	}
 
 	StorageStatus string
+	StorageType   string
 
 	BlobStorageConfig struct {
 		// Status means whether this storage is active for writing.
@@ -50,7 +55,7 @@ type (
 		// StorageId is the id of the external storage, it's used to identify the external storage in the EncodedObject that is stored in the workflow history
 		StorageId string `yaml:"storageId"`
 		// StorageType is the type of the external storage, currently only s3 is supported
-		StorageType string `yaml:"storageType"`
+		StorageType StorageType `yaml:"storageType"`
 		// S3Endpoint is the endpoint of s3 service
 		S3Endpoint string `yaml:"s3Endpoint"`
 		// S3Bucket is the bucket name of the S3 storage

integ/config.go

@@ -36,8 +36,8 @@ func createTestConfig(testCfg IwfServiceTestConfig) config.Config {
 			SupportedStorages: []config.BlobStorageConfig{
 				{
 					Status:      config.StorageStatusActive,
-					StorageId:   "s3",
-					StorageType: "s3",
+					StorageId:   "s3-store-id",
+					StorageType: config.StorageTypeS3,
 					S3Endpoint:  "http://localhost:9000",
 					S3Bucket:    "iwf-test-bucket",
 					S3Region:    "us-east-1",

integ/s3_wf_start_input_test.go

@@ -82,18 +82,6 @@ func doTestWorkflowWithS3StartInput(t *testing.T, backendType service.BackendTyp
 	assertions := assert.New(t)
 
 	_, history := wfHandler.GetTestResult()
-	// assertions.Equalf(map[string]interface{}{
-	// 	"S1_start":  1,
-	// 	"S1_decide": 1,
-	// 	"S1_start_input": iwfidl.EncodedObject{
-	// 		Encoding: iwfidl.PtrString("json"),
-	// 		Data:     iwfidl.PtrString("\"12345678901\""),
-	// 	},
-	// 	"S1_decide_input": iwfidl.EncodedObject{
-	// 		Encoding: iwfidl.PtrString("json"),
-	// 		Data:     iwfidl.PtrString("\"12345678901\""),
-	// 	},
-	// }, history, "headers test fail, %v", history)
 
 	assertions.Equal(history["S1_start_input"], iwfidl.EncodedObject{
 		Encoding: iwfidl.PtrString("json"),

integ/util.go

@@ -3,6 +3,7 @@ package integ
 import (
 	"context"
 	"fmt"
+	"github.com/indeedeng/iwf/service/common/blobstore"
 	"log"
 	"net/http"
 	"testing"
@@ -124,9 +125,10 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 
 		testCfg := createTestConfig(config)
 		s3Client := iwf.CreateS3Client(testCfg, context.Background())
+		store := blobstore.NewBlobStore(s3Client, testNamespace, testCfg.ExternalStorage, logger)
 
 		uclient = temporalapi.NewTemporalClient(temporalClient, testNamespace, dataConverter, config.MemoEncryption, &testCfg.Api.QueryWorkflowFailedRetryPolicy)
-		iwfService := api.NewService(testCfg, uclient, logger, s3Client, "test/") // TODO pass s3 client for integ test
+		iwfService := api.NewService(testCfg, uclient, logger, store)
 		iwfServer := &http.Server{
 			Addr:    ":" + testIwfServerPort,
 			Handler: iwfService,
@@ -138,7 +140,7 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 		}()
 
 		// start iwf interpreter worker
-		interpreter := temporal.NewInterpreterWorker(testCfg, temporalClient, service.TaskQueue, config.MemoEncryption, dataConverter, uclient, s3Client)
+		interpreter := temporal.NewInterpreterWorker(testCfg, temporalClient, service.TaskQueue, config.MemoEncryption, dataConverter, uclient, store)
 		if *disableStickyCache {
 			interpreter.StartWithStickyCacheDisabledForTest()
 		} else {
@@ -163,9 +165,10 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 
 		testCfg := createTestConfig(config)
 		s3Client := iwf.CreateS3Client(testCfg, context.Background())
+		store := blobstore.NewBlobStore(s3Client, iwf.DefaultCadenceDomain, testCfg.ExternalStorage, logger)
 
 		uclient = cadenceapi.NewCadenceClient(iwf.DefaultCadenceDomain, cadenceClient, serviceClient, encoded.GetDefaultDataConverter(), closeFunc, &testCfg.Api.QueryWorkflowFailedRetryPolicy)
-		iwfService := api.NewService(testCfg, uclient, logger, s3Client, "test/") // pass in for integ tests
+		iwfService := api.NewService(testCfg, uclient, logger, store)
 		iwfServer := &http.Server{
 			Addr:    ":" + testIwfServerPort,
 			Handler: iwfService,
@@ -177,7 +180,7 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 		}()
 
 		// start iwf interpreter worker
-		interpreter := cadence.NewInterpreterWorker(testCfg, serviceClient, iwf.DefaultCadenceDomain, service.TaskQueue, closeFunc, uclient, s3Client)
+		interpreter := cadence.NewInterpreterWorker(testCfg, serviceClient, iwf.DefaultCadenceDomain, service.TaskQueue, closeFunc, uclient, store)
 		if *disableStickyCache {
 			interpreter.StartWithStickyCacheDisabledForTest()
 		} else {

service/api/handler.go

@@ -1,8 +1,8 @@
 package api
 
 import (
-	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/indeedeng/iwf/config"
+	"github.com/indeedeng/iwf/service/common/blobstore"
 	"net/http"
 
 	"github.com/indeedeng/iwf/service"
@@ -21,8 +21,8 @@ type handler struct {
 	logger log.Logger
 }
 
-func newHandler(config config.Config, client uclient.UnifiedClient, logger log.Logger, s3Client *s3.Client, s3PathPrefix string) *handler {
-	svc, err := NewApiService(config, client, service.TaskQueue, logger, s3Client, s3PathPrefix)
+func newHandler(config config.Config, client uclient.UnifiedClient, logger log.Logger, store blobstore.BlobStore) *handler {
+	svc, err := NewApiService(config, client, service.TaskQueue, logger, store)
 	if err != nil {
 		panic(err)
 	}

service/api/routers.go

@@ -1,10 +1,10 @@
 package api
 
 import (
-	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/gin-gonic/gin"
 	"github.com/indeedeng/iwf/config"
 	uclient "github.com/indeedeng/iwf/service/client"
+	"github.com/indeedeng/iwf/service/common/blobstore"
 	"github.com/indeedeng/iwf/service/common/log"
 )
 
@@ -29,10 +29,10 @@ const WorkflowRpcApiPath = "/api/v1/workflow/rpc"
 const InfoHealthCheck = "/info/healthcheck"
 
 // NewService returns a new router.
-func NewService(config config.Config, client uclient.UnifiedClient, logger log.Logger, s3Client *s3.Client, s3PathPrefix string) *gin.Engine {
+func NewService(config config.Config, client uclient.UnifiedClient, logger log.Logger, store blobstore.BlobStore) *gin.Engine {
 	router := gin.Default()
 
-	handler := newHandler(config, client, logger, s3Client, s3PathPrefix)
+	handler := newHandler(config, client, logger, store)
 
 	router.GET("/", handler.index)
 	router.POST(WorkflowStartApiPath, handler.apiV1WorkflowStart)

service/api/service.go

@@ -3,13 +3,12 @@ package api
 import (
 	"context"
 	"fmt"
+	"github.com/indeedeng/iwf/service/common/blobstore"
 	"net/http"
 	"os"
 	"strings"
 	"time"
 
-	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/google/uuid"
 	"github.com/indeedeng/iwf/config"
 	"github.com/indeedeng/iwf/service/common/event"
@@ -31,39 +30,26 @@ import (
 )
 
 type serviceImpl struct {
-	client        uclient.UnifiedClient
-	s3Client      *s3.Client
-	s3PathPrefix  string // it's recommended to be the Temporal namespace or Cadence domain + "/"
-	activeStorage *config.BlobStorageConfig
-	taskQueue     string
-	logger        log.Logger
-	config        config.Config
+	client    uclient.UnifiedClient
+	store     blobstore.BlobStore
+	taskQueue string
+	logger    log.Logger
+	config    config.Config
 }
 
 func (s *serviceImpl) Close() {
 	s.client.Close()
 }
 
 func NewApiService(
-	cfg config.Config, client uclient.UnifiedClient, taskQueue string, logger log.Logger, s3Client *s3.Client, s3PathPrefix string,
+	cfg config.Config, client uclient.UnifiedClient, taskQueue string, logger log.Logger, store blobstore.BlobStore,
 ) (ApiService, error) {
-	// get the first active storage
-	var activeStorage *config.BlobStorageConfig
-	for _, storage := range cfg.ExternalStorage.SupportedStorages {
-		if storage.Status == config.StorageStatusActive {
-			activeStorage = &storage
-			break
-		}
-	}
-
 	return &serviceImpl{
-		client:        client,
-		s3Client:      s3Client,
-		s3PathPrefix:  s3PathPrefix,
-		activeStorage: activeStorage,
-		taskQueue:     taskQueue,
-		logger:        logger,
-		config:        cfg,
+		client:    client,
+		store:     store,
+		taskQueue: taskQueue,
+		logger:    logger,
+		config:    cfg,
 	}, nil
 }
 
@@ -172,18 +158,15 @@ func (s *serviceImpl) ApiV1WorkflowStartPost(
 			// 2. if it is, upload the input to S3
 			uuid := uuid.New().String()
 			yyyymmdd := time.Now().Format("20060102")
-			// namespace/yymmdd/workflowId/uuid
-			objectKey := fmt.Sprintf("%s%s/%s/%s", s.s3PathPrefix, yyyymmdd, req.GetWorkflowId(), uuid)
-			err := putObject(ctx, s.s3Client,
-				s.activeStorage.S3Bucket,
-				objectKey,
-				*input.StateInput.Data)
+			// yymmdd/workflowId/uuid
+			objectKey := fmt.Sprintf("%s/%s/%s", yyyymmdd, req.GetWorkflowId(), uuid)
+			storeId, err := s.store.WriteObject(ctx, objectKey, input.StateInput.GetData())
 			if err != nil {
 				return nil, s.handleError(err, WorkflowStartApiPath, req.GetWorkflowId())
 			}
 			// 3. replace the input with the S3 object
 			newStateInput := iwfidl.EncodedObject{
-				ExtStoreId: iwfidl.PtrString(s.activeStorage.StorageId),
+				ExtStoreId: iwfidl.PtrString(storeId),
 				ExtPath:    iwfidl.PtrString(objectKey),
 				Encoding:   iwfidl.PtrString(*input.StateInput.Encoding),
 			}
@@ -225,16 +208,6 @@ func (s *serviceImpl) ApiV1WorkflowStartPost(
 	}, nil
 }
 
-func putObject(ctx context.Context, client *s3.Client, bucketName string, key, content string) error {
-	_, err := client.PutObject(ctx, &s3.PutObjectInput{
-		Bucket:      aws.String(bucketName),
-		Key:         aws.String(key),
-		Body:        strings.NewReader(content),
-		ContentType: aws.String("application/json"),
-	})
-	return err
-}
-
 func overrideWorkflowConfig(configOverride iwfidl.WorkflowConfig, workflowConfig *iwfidl.WorkflowConfig) {
 	if configOverride.ExecutingStateIdMode != nil {
 		workflowConfig.ExecutingStateIdMode = configOverride.ExecutingStateIdMode

service/common/blobstore/interfaces.go

@@ -0,0 +1,11 @@
+package blobstore
+
+import "context"
+
+type BlobStore interface {
+	// WriteObject will write to the current active store
+	// returns the active storeId
+	WriteObject(ctx context.Context, path, data string) (storeId string, err error)
+	// ReadObject will read from the store by storeId
+	ReadObject(ctx context.Context, storeId, path string) (string, error)
+}