Add s3 dependencies and state input of start workflow API (#575)

Author: longquanzheng

cmd/s3-example/main.go

@@ -24,15 +24,20 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	rawLog "log"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsConfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/indeedeng/iwf/config"
 	cadenceapi "github.com/indeedeng/iwf/service/client/cadence"
 	temporalapi "github.com/indeedeng/iwf/service/client/temporal"
 	ggrpc "google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
-	rawLog "log"
-	"strings"
-	"sync"
-	"time"
 
 	isvc "github.com/indeedeng/iwf/service"
 	"github.com/indeedeng/iwf/service/api"
@@ -199,12 +204,17 @@ func launchTemporalService(
 	svcName string, config config.Config, unifiedClient uclient.UnifiedClient, temporalClient client.Client,
 	logger log.Logger,
 ) {
+	s3Client := CreateS3Client(config, context.Background())
 	switch svcName {
 	case serviceAPI:
-		svc := api.NewService(config, unifiedClient, logger.WithTags(tag.Service(svcName)))
+		svc := api.NewService(
+			config, unifiedClient, logger.WithTags(tag.Service(svcName)),
+			s3Client,
+			config.Interpreter.Temporal.Namespace+"/",
+		)
 		rawLog.Fatal(svc.Run(fmt.Sprintf(":%v", config.Api.Port)))
 	case serviceInterpreter:
-		interpreter := temporal.NewInterpreterWorker(config, temporalClient, isvc.TaskQueue, false, nil, unifiedClient)
+		interpreter := temporal.NewInterpreterWorker(config, temporalClient, isvc.TaskQueue, false, nil, unifiedClient, s3Client)
 		interpreter.Start()
 	default:
 		rawLog.Fatalf("Invalid service: %v", svcName)
@@ -220,12 +230,17 @@ func launchCadenceService(
 	closeFunc func(),
 	logger log.Logger,
 ) {
+	s3Client := CreateS3Client(config, context.Background())
 	switch svcName {
 	case serviceAPI:
-		svc := api.NewService(config, unifiedClient, logger.WithTags(tag.Service(svcName)))
+		svc := api.NewService(
+			config, unifiedClient, logger.WithTags(tag.Service(svcName)),
+			s3Client,
+			config.Interpreter.Cadence.Domain+"/",
+		)
 		rawLog.Fatal(svc.Run(fmt.Sprintf(":%v", config.Api.Port)))
 	case serviceInterpreter:
-		interpreter := cadence.NewInterpreterWorker(config, service, domain, isvc.TaskQueue, closeFunc, unifiedClient)
+		interpreter := cadence.NewInterpreterWorker(config, service, domain, isvc.TaskQueue, closeFunc, unifiedClient, s3Client)
 		interpreter.Start()
 	default:
 		rawLog.Fatalf("Invalid service: %v", svcName)
@@ -340,3 +355,78 @@ func newPrometheusScope(c prometheus.Configuration, logger log.Logger) tally.Sco
 	logger.Info("prometheus metrics scope created")
 	return scope
 }
+
+func CreateS3Client(cfg config.Config, ctx context.Context) *s3.Client {
+
+	if !cfg.ExternalStorage.Enabled {
+		return nil
+	}
+
+	// get the first active storage
+	var activeStorage *config.BlobStorageConfig
+	for _, storage := range cfg.ExternalStorage.SupportedStorages {
+		if storage.Status == config.StorageStatusActive {
+			activeStorage = &storage
+			break
+		}
+	}
+	if activeStorage == nil {
+		rawLog.Fatal("no active storage found")
+	}
+
+	if activeStorage.StorageType != "s3" {
+		rawLog.Fatal("only s3 is supported for external storage")
+	}
+
+	// Create custom resolver for MinIO endpoint
+	customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
+
+		if service == s3.ServiceID {
+			return aws.Endpoint{
+				URL:               activeStorage.S3Endpoint,
+				HostnameImmutable: true,
+				Source:            aws.EndpointSourceCustom,
+			}, nil
+		}
+		return aws.Endpoint{}, fmt.Errorf("unknown endpoint requested")
+	})
+
+	// Load AWS config with custom credentials and endpoint
+	cfg2, err := awsConfig.LoadDefaultConfig(ctx,
+		awsConfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(activeStorage.S3AccessKey, activeStorage.S3SecretKey, "")),
+		awsConfig.WithRegion(activeStorage.S3Region),
+		awsConfig.WithEndpointResolverWithOptions(customResolver),
+	)
+	if err != nil {
+		rawLog.Fatal("failed to load AWS config", tag.Error(err))
+	}
+
+	// Create S3 client with path-style addressing (required for MinIO)
+	client := s3.NewFromConfig(cfg2, func(o *s3.Options) {
+		o.UsePathStyle = true
+	})
+
+	createBucketIfNotExists(ctx, client, activeStorage.S3Bucket)
+
+	return client
+}
+
+func createBucketIfNotExists(ctx context.Context, client *s3.Client, bucketName string) {
+	// Check if bucket exists
+	_, err := client.HeadBucket(ctx, &s3.HeadBucketInput{
+		Bucket: aws.String(bucketName),
+	})
+
+	if err != nil {
+		// Bucket doesn't exist, create it
+		_, err = client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket: aws.String(bucketName),
+		})
+		if err != nil {
+			rawLog.Fatal("failed to create bucket", tag.Error(err))
+		}
+		rawLog.Printf("bucket created successfully: %s", bucketName)
+	} else {
+		rawLog.Printf("bucket already exists: %s", bucketName)
+	}
+}

cmd/server/iwf/iwf.go

@@ -2,14 +2,20 @@ package config
 
 import (
 	"fmt"
+	"log"
+	"os"
+	"time"
+
 	"github.com/indeedeng/iwf/gen/iwfidl"
 	"github.com/uber-go/tally/v4/prometheus"
 	temporalWorker "go.temporal.io/sdk/worker"
 	cadenceWorker "go.uber.org/cadence/worker"
 	"gopkg.in/yaml.v3"
-	"log"
-	"os"
-	"time"
+)
+
+const (
+	StorageStatusActive   = "active"
+	StorageStatusInactive = "inactive"
 )
 
 type (
@@ -20,6 +26,41 @@ type (
 		Api ApiConfig `yaml:"api"`
 		// Interpreter is the service behind, either Cadence or Temporal is required
 		Interpreter Interpreter `yaml:"interpreter"`
+		// ExternalStorage is the external storage config
+		ExternalStorage ExternalStorageConfig `yaml:"externalStorage"`
+	}
+
+	ExternalStorageConfig struct {
+		Enabled bool `yaml:"enabled"`
+		// ThresholdInBytes is the size threshold of encodedObject
+		// that will be stored by external storage(picking the current active one)
+		ThresholdInBytes int `yaml:"thresholdInBytes"`
+		// SupportedStorages is the list of supported storage
+		// Only one can be active, meaning the one that will be used for writing.
+		// The non-active ones are for read only.
+		SupportedStorages []BlobStorageConfig `yaml:"supportedStorages"`
+	}
+
+	StorageStatus string
+
+	BlobStorageConfig struct {
+		// Status means whether this storage is active for writing.
+		// Only one of the supported storages can be active
+		Status StorageStatus
+		// 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"`
+		// S3Endpoint is the endpoint of s3 service
+		S3Endpoint string `yaml:"s3Endpoint"`
+		// S3Bucket is the bucket name of the S3 storage
+		S3Bucket string `yaml:"s3Bucket"`
+		// S3Region is the region of the S3 storage
+		S3Region string `yaml:"s3Region"`
+		// S3AccessKey is the access key of the S3 storage
+		S3AccessKey string `yaml:"s3AccessKey"`
+		// S3SecretKey is the secret key of the S3 storage
+		S3SecretKey string `yaml:"s3SecretKey"`
 	}
 
 	ApiConfig struct {

config/config.go

@@ -19,3 +19,15 @@ interpreter:
       timerType: histogram
   defaultWorkflowConfig:
     continueAsNewThreshold: 100
+externalStorage:
+  enabled: true
+  thresholdInBytes: 100000 # 100 KB
+  supportedStorages:
+    - status: active
+      storageId: "local-dev-s3"
+      storageType: "s3"
+      s3Endpoint: "http://localhost:9000"
+      s3Bucket: "iwf-test-bucket"
+      s3Region: "us-east-1"
+      s3AccessKey: "minioadmin"
+      s3SecretKey: "minioadmin"

config/development.yaml

@@ -98,6 +98,24 @@ services:
     volumes:
       - ./init-ci-cadence.sh:/etc/cadence/init-ci-cadence.sh
     entrypoint: sh -c "/etc/cadence/init-ci-cadence.sh"
+  minio:
+    container_name: minio
+    image: minio/minio:${MINIO_VERSION:-latest}
+    ports:
+      - "9000:9000"
+      - "9001:9001"
+    environment:
+      - MINIO_ROOT_USER=minioadmin
+      - MINIO_ROOT_PASSWORD=minioadmin
+      - MINIO_DOMAIN=minio
+    command: server /data --console-address ":9001"
+    networks:
+      - testing-network
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
+      interval: 30s
+      timeout: 20s
+      retries: 3    
 networks:
   testing-network:
     driver: bridge

docker-compose/ci-cadence-dependencies.yml

@@ -147,6 +147,24 @@ services:
     volumes:
       - ./init-ci-cadence.sh:/etc/cadence/init-ci-cadence.sh
     entrypoint: sh -c "/etc/cadence/init-ci-cadence.sh"
+  minio:
+    container_name: minio
+    image: minio/minio:${MINIO_VERSION:-latest}
+    ports:
+      - "9000:9000"
+      - "9001:9001"
+    environment:
+      - MINIO_ROOT_USER=minioadmin
+      - MINIO_ROOT_PASSWORD=minioadmin
+      - MINIO_DOMAIN=minio
+    command: server /data --console-address ":9001"
+    networks:
+      - testing-network
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
+      interval: 30s
+      timeout: 20s
+      retries: 3
 networks:
   testing-network:
     driver: bridge

docker-compose/ci-cadence-temporal-dependencies.yml

@@ -66,6 +66,24 @@ services:
     volumes:
       - ./init-ci-temporal.sh:/etc/temporal/init-ci-temporal.sh
     entrypoint: sh -c "/etc/temporal/init-ci-temporal.sh"
+  minio:
+    container_name: minio
+    image: minio/minio:${MINIO_VERSION:-latest}
+    ports:
+      - "9000:9000"
+      - "9001:9001"
+    environment:
+      - MINIO_ROOT_USER=minioadmin
+      - MINIO_ROOT_PASSWORD=minioadmin
+      - MINIO_DOMAIN=minio
+    command: server /data --console-address ":9001"
+    networks:
+      - testing-network
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
+      interval: 30s
+      timeout: 20s
+      retries: 3    
 networks:
   testing-network:
     driver: bridge

docker-compose/ci-temporal-dependencies.yml

@@ -170,7 +170,27 @@ services:
       - testing-network
     ports:
       - 8233:8080
+  minio:
+    container_name: minio
+    image: minio/minio:${MINIO_VERSION:-latest}
+    ports:
+      - "9000:9000"
+      - "9001:9001"
+    environment:
+      - MINIO_ROOT_USER=minioadmin
+      - MINIO_ROOT_PASSWORD=minioadmin
+      - MINIO_DOMAIN=minio
+    command: server /data --console-address ":9001"
+    networks:
+      - testing-network
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
+      interval: 30s
+      timeout: 20s
+      retries: 3
 networks:
   testing-network:
     driver: bridge
-    name: testing-network
+    name: testing-network
+volumes:
+  minio-data:

docker-compose/integ-dependencies.yml

@@ -80,6 +80,7 @@ Class | Method | HTTP request | Description
 *DefaultApi* | [**ApiV1WorkflowConfigUpdatePost**](docs/DefaultApi.md#apiv1workflowconfigupdatepost) | **Post** /api/v1/workflow/config/update | update the config of a workflow
 *DefaultApi* | [**ApiV1WorkflowDataobjectsGetPost**](docs/DefaultApi.md#apiv1workflowdataobjectsgetpost) | **Post** /api/v1/workflow/dataobjects/get | get workflow data objects aka data attributes
 *DefaultApi* | [**ApiV1WorkflowDataobjectsSetPost**](docs/DefaultApi.md#apiv1workflowdataobjectssetpost) | **Post** /api/v1/workflow/dataobjects/set | set workflow data objects aka data attributes
+*DefaultApi* | [**ApiV1WorkflowEncodedobjectLoadPost**](docs/DefaultApi.md#apiv1workflowencodedobjectloadpost) | **Post** /api/v1/workflow/encodedobject/load | load encoded object from storage
 *DefaultApi* | [**ApiV1WorkflowGetPost**](docs/DefaultApi.md#apiv1workflowgetpost) | **Post** /api/v1/workflow/get | get a workflow's status and results(if completed & requested)
 *DefaultApi* | [**ApiV1WorkflowGetWithWaitPost**](docs/DefaultApi.md#apiv1workflowgetwithwaitpost) | **Post** /api/v1/workflow/getWithWait | get a workflow's status and results(if completed & requested), wait if the workflow is still running
 *DefaultApi* | [**ApiV1WorkflowInternalDumpPost**](docs/DefaultApi.md#apiv1workflowinternaldumppost) | **Post** /api/v1/workflow/internal/dump | dump internal info of a workflow