Add S3 count API for testing verification (#579)

Author: longquanzheng

integ/s3_wf_init_data_attributes_test.go

@@ -76,7 +76,7 @@ func doTestWorkflowWithS3InitDataAttributes(t *testing.T, backendType service.Ba
 
 	wfInput := &iwfidl.EncodedObject{
 		Encoding: iwfidl.PtrString("json"),
-		Data:     iwfidl.PtrString("\"test-input\""),
+		Data:     iwfidl.PtrString("test"),
 	}
 
 	req := apiClient.DefaultApi.ApiV1WorkflowStartPost(context.Background())
@@ -131,4 +131,8 @@ func doTestWorkflowWithS3InitDataAttributes(t *testing.T, backendType service.Ba
 	assertions.Equal(history["S2_decide_attr2_data"], *s3_init_data_attributes.TestDataAttributeVal2.Data, "S2_decide attr2 data should match initial value")
 	assertions.Equal(history["S2_decide_attr3_data"], *s3_init_data_attributes.TestDataAttributeVal3.Data, "S2_decide attr3 data should match initial value")
 	assertions.Equal(history["S2_decide_validation_pass"], true, "S2_decide validation should pass - all data attributes match initial values exactly")
+
+	objectCount, err := globalBlobStore.CountWorkflowObjectsForTesting(context.Background(), wfId)
+	assertions.Nil(err)
+	assertions.Equal(int64(2), objectCount)
 }

integ/s3_wf_start_input_test.go

@@ -94,4 +94,8 @@ func doTestWorkflowWithS3StartInput(t *testing.T, backendType service.BackendTyp
 
 	assertions.Equal(history["S1_start"], int64(1), "S1_start is not equal")
 	assertions.Equal(history["S1_decide"], int64(1), "S1_decide is not equal")
+
+	objectCount, err := globalBlobStore.CountWorkflowObjectsForTesting(context.Background(), wfId)
+	assertions.Nil(err)
+	assertions.Equal(int64(1), objectCount)
 }

integ/util.go

@@ -110,6 +110,9 @@ func startIwfServiceWithClient(backendType service.BackendType) (uclient uclient
 //var integCadenceUclientCached api.UnifiedClient
 //var integTemporalUclientCached api.UnifiedClient
 
+// globalBlobStore is a global var in this package for testing
+var globalBlobStore blobstore.BlobStore
+
 func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.UnifiedClient, closeFunc func()) {
 	if config.BackendType == service.BackendTypeTemporal {
 		dataConverter := converter.GetDefaultDataConverter()
@@ -125,10 +128,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, client.MetricsNopHandler)
+		globalBlobStore = blobstore.NewBlobStore(s3Client, testNamespace, testCfg.ExternalStorage, logger, client.MetricsNopHandler)
 
 		uclient = temporalapi.NewTemporalClient(temporalClient, testNamespace, dataConverter, config.MemoEncryption, &testCfg.Api.QueryWorkflowFailedRetryPolicy)
-		iwfService := api.NewService(testCfg, uclient, logger, store)
+		iwfService := api.NewService(testCfg, uclient, logger, globalBlobStore)
 		iwfServer := &http.Server{
 			Addr:    ":" + testIwfServerPort,
 			Handler: iwfService,
@@ -140,7 +143,7 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 		}()
 
 		// start iwf interpreter worker
-		interpreter := temporal.NewInterpreterWorker(testCfg, temporalClient, service.TaskQueue, config.MemoEncryption, dataConverter, uclient, store)
+		interpreter := temporal.NewInterpreterWorker(testCfg, temporalClient, service.TaskQueue, config.MemoEncryption, dataConverter, uclient, globalBlobStore)
 		if *disableStickyCache {
 			interpreter.StartWithStickyCacheDisabledForTest()
 		} else {
@@ -165,10 +168,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, client.MetricsNopHandler)
+		globalBlobStore = blobstore.NewBlobStore(s3Client, iwf.DefaultCadenceDomain, testCfg.ExternalStorage, logger, client.MetricsNopHandler)
 
 		uclient = cadenceapi.NewCadenceClient(iwf.DefaultCadenceDomain, cadenceClient, serviceClient, encoded.GetDefaultDataConverter(), closeFunc, &testCfg.Api.QueryWorkflowFailedRetryPolicy)
-		iwfService := api.NewService(testCfg, uclient, logger, store)
+		iwfService := api.NewService(testCfg, uclient, logger, globalBlobStore)
 		iwfServer := &http.Server{
 			Addr:    ":" + testIwfServerPort,
 			Handler: iwfService,
@@ -180,7 +183,7 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 		}()
 
 		// start iwf interpreter worker
-		interpreter := cadence.NewInterpreterWorker(testCfg, serviceClient, iwf.DefaultCadenceDomain, service.TaskQueue, closeFunc, uclient, store)
+		interpreter := cadence.NewInterpreterWorker(testCfg, serviceClient, iwf.DefaultCadenceDomain, service.TaskQueue, closeFunc, uclient, globalBlobStore)
 		if *disableStickyCache {
 			interpreter.StartWithStickyCacheDisabledForTest()
 		} else {

service/common/blobstore/interfaces.go

@@ -27,6 +27,12 @@ type BlobStore interface {
 	DeleteObjectPath(ctx context.Context, storeId, path string) error
 	// ListObjectPaths will list the paths of yyyymmdd$workflowId
 	ListObjectPaths(ctx context.Context, input ListObjectPathsInput) (*ListObjectPathsOutput, error)
+	// CountWorkflowObjectsForTesting is for testing ONLY.
+	// count the number of S3 objects for this workflowId
+	// Limitation:
+	//  1. It doesn't count across two days(so expect test to fail if you happen to run the test across day boundary :)
+	//  2. Only count less than 1000 objects(because it only make one API call to S3 which return at most 1000 objects)
+	CountWorkflowObjectsForTesting(ctx context.Context, workflowId string) (int64, error)
 }
 
 type ListObjectPathsInput struct {

service/common/blobstore/store_impl.go

@@ -5,15 +5,16 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
+	"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/log"
 	"go.temporal.io/sdk/client"
-	"io"
-	"strings"
-	"time"
 )
 
 type blobStoreImpl struct {
@@ -137,6 +138,23 @@ func getObject(ctx context.Context, client *s3.Client, bucketName, key string) (
 	return buf.String(), nil
 }
 
+func (b *blobStoreImpl) CountWorkflowObjectsForTesting(ctx context.Context, workflowId string) (int64, error) {
+	// Create the prefix to match objects for this workflowId for today
+	yyyymmdd := time.Now().Format("20060102")
+	prefix := fmt.Sprintf("%s%s$%s/", b.pathPrefix, yyyymmdd, workflowId)
+
+	// List objects with the prefix (limited to 1000 objects as documented)
+	result, err := b.s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+		Bucket: aws.String(b.activeStorage.S3Bucket),
+		Prefix: aws.String(prefix),
+	})
+	if err != nil {
+		return 0, err
+	}
+
+	return int64(len(result.Contents)), nil
+}
+
 func (b *blobStoreImpl) DeleteObjectPath(ctx context.Context, storeId, path string) error {
 	//TODO implement me
 	panic("implement me")