Skip to main content

External Storage - Go SDK

View Markdown

The Temporal Service enforces a 2 MB per-payload limit by default. This limit is configurable on self-hosted deployments. When your Workflows or Activities handle data larger than the limit, you can offload payloads to external storage, such as Amazon S3, and pass a small reference token through the Event History instead. This page shows you how to set up External Storage with Amazon S3 or Google Cloud Storage, and how to implement a custom storage driver.

For a conceptual overview of External Storage and its use cases, see External Storage.

Store and retrieve large payloads with Amazon S3 or Google Cloud Storage​

The Go SDK includes storage drivers for Amazon S3 and Google Cloud Storage. Select your storage backend in the tabs that follow. Only the driver setup differs between the two. Everything after that is the same.

Prerequisites​

  • A bucket that you have read and write access to. Refer to lifecycle management to ensure that your payloads remain available for the entire lifetime of the Workflow. For multi-region durability, see Durable External Storage.

  • Bucket credentials on both your Temporal Client and your Workers, since each reaches the bucket directly. Refer to Amazon S3 access control or Cloud Storage IAM permissions for how to grant access.

  • Permission to write objects on processes that store payloads, and to read objects on processes that retrieve them. Storing also requires read permission, because the drivers check whether an object already exists before uploading.

  • Install the driver module, the client adapter for your cloud provider's SDK, and that SDK:

    go get go.temporal.io/sdk/contrib/aws/s3driver \
      go.temporal.io/sdk/contrib/aws/s3driver/awssdkv2 \
      github.com/aws/aws-sdk-go-v2/config \
      github.com/aws/aws-sdk-go-v2/service/s3

Procedure​

  1. Create a storage client, wrap it in the matching driver client, and pass the result to the driver. Each cloud SDK picks up your standard credentials from the environment:

    The AWS SDK reads environment variables, an IAM role, or your AWS config file.

    features/snippets/external_storage/s3_setup/s3_driver_create.go

    cfg, err := config.LoadDefaultConfig(context.Background(),
    	config.WithRegion("us-east-2"),
    )
    if err != nil {
    	log.Fatalf("load AWS config: %v", err)
    }

    driver, err := s3driver.NewDriver(s3driver.Options{
    	Client: awssdkv2.NewClient(s3.NewFromConfig(cfg)),
    	Bucket: s3driver.StaticBucket("my-temporal-payloads"),
    })
    if err != nil {
    	log.Fatalf("create S3 driver: %v", err)
    }

    To route payloads to different buckets at runtime, pass a BucketFunc as Bucket instead of using StaticBucket. The function receives the store context and the payload, and returns a bucket name.

  2. Configure the driver on ExternalStorage and pass it in your Client options. This step is the same for every driver:

    features/snippets/external_storage/s3_setup/s3_external_storage_setup.go

    c, err := client.Dial(client.Options{
    	HostPort: "localhost:7233",
    	ExternalStorage: converter.ExternalStorage{
    		Drivers: []converter.StorageDriver{driver},
    	},
    })
    if err != nil {
    	log.Fatalf("connect to Temporal: %v", err)
    }
    defer c.Close()

    w := worker.New(c, "my-task-queue", worker.Options{})

    A Worker inherits this configuration from the Client it is created with. When your Workers run in their own process, repeat this setup there.

    By default, payloads of 256 KiB or larger are offloaded to external storage. You can adjust this with the PayloadSizeThreshold option, even setting it to 1 to externalize all payloads regardless of size. Refer to Configure payload size threshold for more information.

    All Workflows and Activities running on the Worker use the storage driver automatically without changes to your business logic. Both drivers upload and download payloads concurrently, and verify a SHA-256 hash of the contents on retrieve. Each driver rejects any single payload larger than MaxPayloadSize, which defaults to 50 MiB.

    The hash forms part of the object key, along with the Namespace, Workflow Id, and Run Id. When one Workflow Run passes the same payload to several Activities, the key is identical each time, so the driver uploads the payload once and the later writes reuse that object. A different Run, Workflow, or Namespace produces a different key, so it stores its own copy even when the bytes are identical. Storage therefore scales with the number of Runs, not with how many times a Run passes a payload around.

    The S3 driver includes diagnostic metadata, such as the AWS region, in error messages to help troubleshoot storage failures.

Implement a custom storage driver​

If you need a storage backend other than what the built-in drivers allow, you can implement your own storage driver. Refer to Choose a storage system for guidance on selecting a backing store and Lifecycle management for retention requirements.

The following example shows a custom driver that uses local disk as the backing store. This example is for local development and testing only. In production, use a durable storage system that is accessible to all Workers:

features/snippets/external_storage/custom_driver/custom_storage_driver.go

type LocalDiskStorageDriver struct {
	storeDir string
}

func NewLocalDiskStorageDriver(storeDir string) converter.StorageDriver {
	return &LocalDiskStorageDriver{storeDir: storeDir}
}

func (d *LocalDiskStorageDriver) Name() string {
	return "my-local-disk"
}

func (d *LocalDiskStorageDriver) Type() string {
	return "local-disk"
}

func (d *LocalDiskStorageDriver) Store(
	ctx converter.StorageDriverStoreContext,
	payloads []*commonpb.Payload,
) ([]converter.StorageDriverClaim, error) {
	dir := d.storeDir
	switch info := ctx.Target.(type) {
	case converter.StorageDriverWorkflowInfo:
		if info.WorkflowID != "" {
			dir = filepath.Join(d.storeDir, info.Namespace, info.WorkflowID)
		}
	case converter.StorageDriverActivityInfo:
		// StorageDriverActivityInfo is only used for standalone (non-workflow-bound)
		// activities. Activities started by a workflow use StorageDriverWorkflowInfo.
		if info.ActivityID != "" {
			dir = filepath.Join(d.storeDir, info.Namespace, info.ActivityID)
		}
	}
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return nil, fmt.Errorf("create store directory: %w", err)
	}

	claims := make([]converter.StorageDriverClaim, len(payloads))
	for i, payload := range payloads {
		key := uuid.NewString() + ".bin"
		filePath := filepath.Join(dir, key)

		data, err := proto.Marshal(payload)
		if err != nil {
			return nil, fmt.Errorf("marshal payload: %w", err)
		}
		if err := os.WriteFile(filePath, data, 0o644); err != nil {
			return nil, fmt.Errorf("write payload: %w", err)
		}

		claims[i] = converter.StorageDriverClaim{
			ClaimData: map[string]string{"path": filePath},
		}
	}
	return claims, nil
}

func (d *LocalDiskStorageDriver) Retrieve(
	ctx converter.StorageDriverRetrieveContext,
	claims []converter.StorageDriverClaim,
) ([]*commonpb.Payload, error) {
	payloads := make([]*commonpb.Payload, len(claims))
	for i, claim := range claims {
		filePath := claim.ClaimData["path"]
		data, err := os.ReadFile(filePath)
		if err != nil {
			return nil, fmt.Errorf("read payload: %w", err)
		}
		payload := &commonpb.Payload{}
		if err := proto.Unmarshal(data, payload); err != nil {
			return nil, fmt.Errorf("unmarshal payload: %w", err)
		}
		payloads[i] = payload
	}
	return payloads, nil
}

The following sections walk through the key parts of the driver implementation.

1. Implement the StorageDriver interface​

A custom driver implements the converter.StorageDriver interface with four methods:

  • Name() returns a unique string that identifies the driver instance. The SDK stores this name in the claim check reference so it can route retrieval requests to the correct driver. Changing the name after payloads have been stored breaks retrieval. For example, two S3 drivers could be named "s3-primary" and "s3-archive".
  • Type() returns a string that identifies the driver implementation. Unlike Name(), this must be the same across all instances of the same driver type regardless of configuration. Two S3 drivers named "s3-primary" and "s3-archive" would both return "aws.s3driver" as their type, while the local disk driver in the custom driver code sample returns "local-disk".
  • Store() receives a slice of payloads and returns one StorageDriverClaim per payload. A claim is a set of string key-value pairs that the driver uses to locate the payload later.
  • Retrieve() receives the claims that Store() produced and returns the original payloads.

2. Store payloads​

In Store(), marshal each Payload protobuf message to bytes with proto.Marshal(payload) and write the bytes to your storage system. The application data has already been serialized by the Payload Converter and Payload Codec before it reaches the driver. See the data conversion pipeline for more details.

Return a StorageDriverClaim for each payload with enough information to retrieve it later. The ctx.Target provides identity information (namespace, Workflow ID) depending on the operation. Use a type switch on StorageDriverWorkflowInfo and StorageDriverActivityInfo to access the concrete values. Consider structuring your storage keys to include this information so that you can identify which Workflow owns each payload.

3. Retrieve payloads​

In Retrieve(), download the bytes using the claim data, then reconstruct the Payload protobuf message with proto.Unmarshal(data, payload). The Payload Converter handles deserializing the application data after the driver returns the payload.

4. Configure the Client​

Pass an ExternalStorage struct with your driver in the Client options:

c, err := client.Dial(client.Options{
    ExternalStorage: converter.ExternalStorage{
        Drivers: []converter.StorageDriver{NewLocalDiskStorageDriver("/tmp/temporal-payload-store")},
    },
})

You can also package your driver as a plugin for easier reuse across services.

Configure payload size threshold​

You can configure the payload size threshold that triggers external storage. By default, payloads of 256 KiB or larger are offloaded to external storage. You can adjust this with the PayloadSizeThreshold option, or set it to 1 to externalize all payloads regardless of size. A value of 0 is interpreted as the default (256 KiB). Payloads smaller than the threshold stay inline in Event History. The size compared against the threshold is that of the serialized Payload, which includes its metadata, not just your data.

features/snippets/external_storage/threshold/threshold_config.go

c, err := client.Dial(client.Options{
	ExternalStorage: converter.ExternalStorage{
		Drivers:              []converter.StorageDriver{driver},
		PayloadSizeThreshold: 1,
	},
})

Use multiple storage drivers​

When you register multiple drivers, you must provide a DriverSelector that implements the StorageDriverSelector interface. The selector chooses which driver stores each payload. Any driver in the list that is not selected for storing is still available for retrieval, which is useful when migrating between storage backends. Return nil from the selector to keep a specific payload inline in Event History.

Multiple drivers are useful in scenarios such as:

  • Driver migration. Your Worker needs to retrieve payloads created by clients that use a different driver than the one you prefer. Register both drivers and use the selector to always pick your preferred driver for new payloads. The old driver remains available for retrieving existing claims.
  • Multi-cloud storage. Route payloads to different storage backends based on your cloud environment. For example, use S3 for Workers running on AWS and GCS for Workers running on Google Cloud. The selector chooses the appropriate driver based on the runtime environment.

Every registered driver needs a distinct name. The S3 and GCS drivers default to "aws.s3driver" and "gcp.gcsdriver", so you can register one of each without extra configuration. Registering two drivers of the same type requires setting the DriverName option on at least one of them.

The following example registers two drivers but always selects preferredDriver for new payloads. The legacyDriver is only registered so the Worker can retrieve payloads that were previously stored with it:

features/snippets/external_storage/multiple_drivers/multiple_drivers.go

type PreferredSelector struct {
	preferred converter.StorageDriver
}

func (s *PreferredSelector) SelectDriver(
	ctx converter.StorageDriverStoreContext,
	payload *commonpb.Payload,
) (converter.StorageDriver, error) {
	return s.preferred, nil
}

func MultipleDriversSetup(preferredDriver, legacyDriver converter.StorageDriver) converter.ExternalStorage {
	return converter.ExternalStorage{
		Drivers:        []converter.StorageDriver{preferredDriver, legacyDriver},
		DriverSelector: &PreferredSelector{preferred: preferredDriver},
	}
}

Multi-region durability with Amazon S3​

To make your S3-backed External Storage tolerant of regional failures, configure the AWS side with Cross-Region Replication (CRR) and an S3 Multi-Region Access Point (MRAP), then point the driver at the MRAP ARN instead of a bucket name. See Durable External Storage for the full pattern and trade-offs.

In code, the only change is the value you pass to s3driver.StaticBucket:

driver, err := s3driver.NewDriver(s3driver.Options{
	Client: awssdkv2.NewClient(s3.NewFromConfig(cfg)),
	Bucket: s3driver.StaticBucket("arn:aws:s3::123456789012:accesspoint/mfzwi23gnjvgw.mrap"),
})

The AWS SDK for Go v2 automatically uses SigV4A signing when the bucket value is an MRAP ARN, so no additional client configuration is required.