s3store

README

Amazon S3 blobstore

Configuration

See https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html#specifying-credentials on how to set up authentication against s3

Enabling archival is done by using the configuration below. Region and bucket URI are required

archival:
  history:
    state: "enabled"
    enableRead: true
    provider:
      s3store:
        region: "us-east-1"
        logLevel: 0
  visibility:
    state: "enabled"
    enableRead: true
    provider:
      s3store:
        region: "us-east-1"
        logLevel: 0

namespaceDefaults:
  archival:
    history:
      state: "enabled"
      URI: "s3://<bucket-name>"
    visibility:
      state: "enabled"
      URI: "s3://<bucket-name>"

Visibility query syntax

You can query the visibility store by using the tctl workflow listarchived command

The syntax for the query is based on SQL

Supported column names are

• WorkflowId String
• WorkflowTypeName String
• StartTime Date
• CloseTime Date
• SearchPrecision String - Day, Hour, Minute, Second

WorkflowId or WorkflowTypeName is required. If filtering on date use StartTime or CloseTime in combination with SearchPrecision.

Searching for a record will be done in times in the UTC timezone

SearchPrecision specifies what range you want to search for records. If you use SearchPrecision = 'Day' it will search all records starting from 2020-01-21T00:00:00Z to 2020-01-21T59:59:59Z

Limitations

• The only operator supported is = due to how records are stored in s3.

Example

Searches for all records done in day 2020-01-21 with the specified workflow id

./tctl --ns samples-namespace workflow listarchived -q "StartTime = '2020-01-21T00:00:00Z' AND WorkflowId='workflow-id' AND SearchPrecision='Day'"

Storage in S3

Workflow runs are stored in s3 using the following structure

s3://<bucket-name>/<namespace-id>/
	history/<workflow-id>/<run-id>
	visibility/
            workflowTypeName/<workflow-type-name>/
                startTimeout/2020-01-21T16:16:11Z/<run-id>
                closeTimeout/2020-01-21T16:16:11Z/<run-id>
            workflowID/<workflow-id>/
                startTimeout/2020-01-21T16:16:11Z/<run-id>
                closeTimeout/2020-01-21T16:16:11Z/<run-id>

Enable AWS SDK Logging with config parameter logLevel. For example enable debug logging with logLevel: 4096. Possbile Values:

• LogOff = 0 = 0x0
• LogDebug = 4096 = 0x1000
• LogDebugWithSigning = 4097 = 0x1001
• LogDebugWithHTTPBody = 4098 = 0x1002
• LogDebugWithRequestRetries = 4100 = 0x1004
• LogDebugWithRequestErrors = 4104 = 0x1008
• LogDebugWithEventStreamBody = 4112 = 0x1010
• LogDebugWithDeprecated = 4128 = 0x1020

Permissions

Your s3 user must have at least the following permissions:

• s3:ListBucket
• s3:GetObject
• s3:PutObject

Using localstack for local development

1. Install awscli from here
2. Install localstack from here
3. Launch localstack with SERVICES=s3 localstack start
4. Create a bucket using aws --endpoint-url=http://localhost:4566 s3 mb s3://temporal-development
5. Launch the server with the localstack s3 environment config--env development-cass-s3 start

Documentation

Overview

Package s3store is a generated GoMock package.

Index

• Constants
• func BucketExists(ctx context.Context, s3cli s3iface.S3API, URI archiver.URI) error
• func Download(ctx context.Context, s3cli s3iface.S3API, URI archiver.URI, key string) ([]byte, error)
• func Encode(message proto.Message) ([]byte, error)
• func IsNotFoundError(err error) bool
• func KeyExists(ctx context.Context, s3cli s3iface.S3API, URI archiver.URI, key string) (bool, error)
• func NewHistoryArchiver(container *archiver.HistoryBootstrapContainer, config *config.S3Archiver) (archiver.HistoryArchiver, error)
• func NewVisibilityArchiver(container *archiver.VisibilityBootstrapContainer, config *config.S3Archiver) (archiver.VisibilityArchiver, error)
• func SerializeToken(token interface{}) ([]byte, error)
• func SoftValidateURI(URI archiver.URI) error
• func Upload(ctx context.Context, s3cli s3iface.S3API, URI archiver.URI, key string, ...) error
• type MockQueryParser
  • func NewMockQueryParser(ctrl *gomock.Controller) *MockQueryParser
  • func (m *MockQueryParser) EXPECT() *MockQueryParserMockRecorder
  • func (m *MockQueryParser) Parse(query string) (*parsedQuery, error)
• type MockQueryParserMockRecorder
  • func (mr *MockQueryParserMockRecorder) Parse(query any) *gomock.Call
• type QueryParser
  • func NewQueryParser() QueryParser

Constants

const (
	WorkflowTypeName = "WorkflowTypeName"
	WorkflowID       = "WorkflowId"
	StartTime        = "StartTime"
	CloseTime        = "CloseTime"
	SearchPrecision  = "SearchPrecision"
)

All allowed fields for filtering

const (
	PrecisionDay    = "Day"
	PrecisionHour   = "Hour"
	PrecisionMinute = "Minute"
	PrecisionSecond = "Second"
)

Precision specific values

const (
	// URIScheme is the scheme for the s3 implementation
	URIScheme = "s3"
)

Functions

func BucketExists

func BucketExists(ctx context.Context, s3cli s3iface.S3API, URI archiver.URI) error

func Download

func Download(ctx context.Context, s3cli s3iface.S3API, URI archiver.URI, key string) ([]byte, error)

func Encode

func Encode(message proto.Message) ([]byte, error)

func IsNotFoundError

func IsNotFoundError(err error) bool

func KeyExists

func KeyExists(ctx context.Context, s3cli s3iface.S3API, URI archiver.URI, key string) (bool, error)

func NewHistoryArchiver

func NewHistoryArchiver(
	container *archiver.HistoryBootstrapContainer,
	config *config.S3Archiver,
) (archiver.HistoryArchiver, error)

NewHistoryArchiver creates a new archiver.HistoryArchiver based on s3

func NewVisibilityArchiver

func NewVisibilityArchiver(
	container *archiver.VisibilityBootstrapContainer,
	config *config.S3Archiver,
) (archiver.VisibilityArchiver, error)

NewVisibilityArchiver creates a new archiver.VisibilityArchiver based on s3

func SerializeToken

func SerializeToken(token interface{}) ([]byte, error)

func SoftValidateURI

func SoftValidateURI(URI archiver.URI) error

Only validates the scheme and buckets are passed

func Upload

func Upload(ctx context.Context, s3cli s3iface.S3API, URI archiver.URI, key string, data []byte) error

Types

type MockQueryParser

type MockQueryParser struct {
	// contains filtered or unexported fields
}

MockQueryParser is a mock of QueryParser interface.

func NewMockQueryParser

func NewMockQueryParser(ctrl *gomock.Controller) *MockQueryParser

NewMockQueryParser creates a new mock instance.

func (*MockQueryParser) EXPECT

func (m *MockQueryParser) EXPECT() *MockQueryParserMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockQueryParser) Parse

func (m *MockQueryParser) Parse(query string) (*parsedQuery, error)

Parse mocks base method.

type MockQueryParserMockRecorder

type MockQueryParserMockRecorder struct {
	// contains filtered or unexported fields
}

MockQueryParserMockRecorder is the mock recorder for MockQueryParser.

func (*MockQueryParserMockRecorder) Parse

func (mr *MockQueryParserMockRecorder) Parse(query any) *gomock.Call

Parse indicates an expected call of Parse.

type QueryParser

type QueryParser interface {
	Parse(query string) (*parsedQuery, error)
}

QueryParser parses a limited SQL where clause into a struct

func NewQueryParser

func NewQueryParser() QueryParser

NewQueryParser creates a new query parser for filestore