Add server option for customizing archiver implementation (#9090)

## What changed?
- Add server option for injecting customHistory/VisibilityArchiverFactory

## Why?
- Add community user to use different archiver implementations

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [x] added new unit test(s)
- [x] added new functional test(s)
This commit is contained in:
Yichao Yang
2026-03-24 00:48:30 +00:00
committed by GitHub
parent 3bb512cdba
commit a45584605e
12 changed files with 1240 additions and 116 deletions
+129 -1
View File
@@ -2,7 +2,15 @@
This README explains how to add new Archiver implementations.
## Steps
There are two approaches:
1. **Built-in implementation** — add the archiver directly to this repository (e.g., `filestore`, `gcloud`, `s3store`). **We are not currently accepting contributions for new built-in archiver implementations.** Maintaining a growing set of built-in implementations places an ongoing maintenance burden on the team, so new implementations should use Option 2 instead.
2. **Custom implementation via server option** — implement the archiver in an external package and inject it into the server at startup using `WithCustomHistoryArchiverFactory` / `WithCustomVisibilityArchiverFactory`. This is the recommended approach for all new archiver implementations.
---
## Option 1: Built-in implementation (in-repo)
**Step 1: Create a new package for your implementation**
@@ -73,6 +81,120 @@ Modify the `./provider/provider.go` file so that the `ArchiverProvider` knows ho
Also, add configs for you archiver to static yaml config files and modify the `HistoryArchiverProvider`
and `VisibilityArchiverProvider` struct in the `../common/service/config.go` accordingly.
---
## Option 2: Custom implementation via server option (external package)
This approach lets you define archiver implementations in your own codebase and inject them into the Temporal server at startup, without modifying the server source.
**Step 1: Implement the HistoryArchiver and VisibilityArchiver interfaces**
Same interfaces as Steps 2 and 3 above.
**Step 2: Implement CustomHistoryArchiverFactory and/or CustomVisibilityArchiverFactory**
```go
// CustomHistoryArchiverFactory constructs a history archiver for a given URI scheme.
// Return provider.ErrUnknownScheme to fall back to the built-in implementation for that scheme.
// If a non-nil archiver is returned, it takes precedence over built-in archiver implementations.
type CustomHistoryArchiverFactory interface {
NewCustomHistoryArchiver(provider.NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error)
}
// CustomVisibilityArchiverFactory constructs a visibility archiver for a given URI scheme.
// Return provider.ErrUnknownScheme to fall back to the built-in implementation for that scheme.
// If a non-nil archiver is returned, it takes precedence over built-in archiver implementations.
type CustomVisibilityArchiverFactory interface {
NewCustomVisibilityArchiver(provider.NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error)
}
```
The params structs provide everything your factory needs to construct an archiver:
```go
type NewCustomHistoryArchiverParams struct {
Scheme string
ExecutionManager persistence.ExecutionManager
Logger log.Logger
MetricsHandler metrics.Handler
Configs map[string]any // from archival.history.provider.customStores.<scheme> in config yaml
}
type NewCustomVisibilityArchiverParams struct {
Scheme string
Logger log.Logger
MetricsHandler metrics.Handler
Configs map[string]any // from archival.visibility.provider.customStores.<scheme> in config yaml
}
```
Example factory implementation using the functional adapter types:
```go
historyFactory := provider.CustomHistoryArchiverFactoryFunc(func(params provider.NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error) {
if params.Scheme != "myscheme" {
return nil, provider.ErrUnknownScheme
}
return mypackage.NewHistoryArchiver(params.ExecutionManager, params.Logger, params.MetricsHandler, params.Configs)
})
visibilityFactory := provider.CustomVisibilityArchiverFactoryFunc(func(params provider.NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error) {
if params.Scheme != "myscheme" {
return nil, provider.ErrUnknownScheme
}
return mypackage.NewVisibilityArchiver(params.Logger, params.MetricsHandler, params.Configs)
})
```
**Step 3: Register the factories with the server**
Pass the factories as server options when constructing the Temporal server:
```go
s, err := temporal.NewServer(
temporal.WithConfig(cfg),
temporal.WithCustomHistoryArchiverFactory(historyFactory),
temporal.WithCustomVisibilityArchiverFactory(visibilityFactory),
// ... other options
)
```
**Step 4: Configure archival in your YAML config**
Enable archival and configure the URI scheme for your implementation. Use `customStores` to pass arbitrary config key-values to your factory:
```yaml
archival:
history:
state: "enabled"
enableRead: true
provider:
customStores:
myscheme: # must match the scheme in your URIs
endpoint: "https://my-storage.example.com"
bucketName: "temporal-history"
visibility:
state: "enabled"
enableRead: true
provider:
customStores:
myscheme:
endpoint: "https://my-storage.example.com"
bucketName: "temporal-visibility"
namespaceDefaults:
archival:
history:
state: "enabled"
URI: "myscheme://temporal-history"
visibility:
state: "enabled"
URI: "myscheme://temporal-visibility"
```
The `customStores.<scheme>` map is passed as `Configs` in the params to your factory. Built-in schemes (`filestore`, `gstorage`, `s3store`) continue to use their own config sections unless your factory handles them (see FAQ below).
---
## FAQ
**If my Archive method can automatically be retried by caller how can I record and access progress between retries?**
@@ -134,6 +256,12 @@ Its usage is simpler than `ExecutionManager`, so archiver implementations can ch
See the `historyIterator.go` file for more details.
Sample usage can be found in the filestore historyArchiver implementation.
**Can a custom factory override a built-in scheme like `filestore`?**
Yes. The custom factory is always consulted first. If your factory returns a non-nil archiver for a scheme that is also built-in (e.g., `filestore`), your implementation takes precedence and the built-in one is never used. Only return `ErrUnknownScheme` for schemes you want to delegate to the built-in implementations.
Note that when overriding a built-in scheme, the `Configs` field in the params is populated from `customStores.<scheme>` — not from the built-in config section (e.g., `filestore:`). If you need those config values, read them from `customStores` instead.
**Should my archiver define all its own error types?**
Each archiver is free to define and return any errors it wants. However many common errors which
+130 -39
View File
@@ -31,12 +31,50 @@ type (
GetVisibilityArchiver(scheme string) (archiver.VisibilityArchiver, error)
}
// NewCustomHistoryArchiverParams provides dependencies for constructing a history archiver.
NewCustomHistoryArchiverParams struct {
Scheme string
ExecutionManager persistence.ExecutionManager
Logger log.Logger
MetricsHandler metrics.Handler
Configs map[string]any
}
// NewCustomVisibilityArchiverParams provides dependencies for constructing a visibility archiver.
NewCustomVisibilityArchiverParams struct {
Scheme string
Logger log.Logger
MetricsHandler metrics.Handler
Configs map[string]any
}
// CustomHistoryArchiverFactory constructs a history archiver for the given scheme.
// Return ErrUnknownScheme to fall back to the default implementation.
// If a non-nil archiver is returned, it takes precedence over built-in archiver implementations.
CustomHistoryArchiverFactory interface {
NewCustomHistoryArchiver(NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error)
}
// CustomVisibilityArchiverFactory constructs a visibility archiver for the given scheme.
// Return ErrUnknownScheme to fall back to the default implementation.
// If a non-nil archiver is returned, it takes precedence over built-in archiver implementations.
CustomVisibilityArchiverFactory interface {
NewCustomVisibilityArchiver(NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error)
}
CustomHistoryArchiverFactoryFunc func(NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error)
CustomVisibilityArchiverFactoryFunc func(NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error)
archiverProvider struct {
sync.RWMutex
historyArchiverConfigs *config.HistoryArchiverProvider
visibilityArchiverConfigs *config.VisibilityArchiverProvider
customHistoryArchiverFactory CustomHistoryArchiverFactory
customVisibilityArchiverFactory CustomVisibilityArchiverFactory
executionManager persistence.ExecutionManager
logger log.Logger
metricsHandler metrics.Handler
@@ -47,22 +85,38 @@ type (
}
)
func (f CustomHistoryArchiverFactoryFunc) NewCustomHistoryArchiver(
params NewCustomHistoryArchiverParams,
) (archiver.HistoryArchiver, error) {
return f(params)
}
func (f CustomVisibilityArchiverFactoryFunc) NewCustomVisibilityArchiver(
params NewCustomVisibilityArchiverParams,
) (archiver.VisibilityArchiver, error) {
return f(params)
}
// NewArchiverProvider returns a new Archiver provider
func NewArchiverProvider(
historyArchiverConfigs *config.HistoryArchiverProvider,
visibilityArchiverConfigs *config.VisibilityArchiverProvider,
customHistoryArchiverFactory CustomHistoryArchiverFactory,
customVisibilityArchiverFactory CustomVisibilityArchiverFactory,
executionManager persistence.ExecutionManager,
logger log.Logger,
metricsHandler metrics.Handler,
) ArchiverProvider {
return &archiverProvider{
historyArchiverConfigs: historyArchiverConfigs,
visibilityArchiverConfigs: visibilityArchiverConfigs,
executionManager: executionManager,
logger: logger,
metricsHandler: metricsHandler,
historyArchivers: make(map[string]archiver.HistoryArchiver),
visibilityArchivers: make(map[string]archiver.VisibilityArchiver),
historyArchiverConfigs: historyArchiverConfigs,
visibilityArchiverConfigs: visibilityArchiverConfigs,
executionManager: executionManager,
logger: logger,
metricsHandler: metricsHandler,
customHistoryArchiverFactory: customHistoryArchiverFactory,
customVisibilityArchiverFactory: customVisibilityArchiverFactory,
historyArchivers: make(map[string]archiver.HistoryArchiver),
visibilityArchivers: make(map[string]archiver.VisibilityArchiver),
}
}
@@ -74,27 +128,46 @@ func (p *archiverProvider) GetHistoryArchiver(scheme string) (historyArchiver ar
}
p.RUnlock()
switch scheme {
case filestore.URIScheme:
if p.historyArchiverConfigs.Filestore == nil {
return nil, ErrArchiverConfigNotFound
if p.customHistoryArchiverFactory != nil {
var customConfigs map[string]any
if p.historyArchiverConfigs != nil {
customConfigs = p.historyArchiverConfigs.CustomStores[scheme]
}
historyArchiver, err = filestore.NewHistoryArchiver(p.executionManager, p.logger, p.metricsHandler, p.historyArchiverConfigs.Filestore)
case gcloud.URIScheme:
if p.historyArchiverConfigs.Gstorage == nil {
return nil, ErrArchiverConfigNotFound
historyArchiver, err = p.customHistoryArchiverFactory.NewCustomHistoryArchiver(NewCustomHistoryArchiverParams{
Scheme: scheme,
ExecutionManager: p.executionManager,
Logger: p.logger,
MetricsHandler: p.metricsHandler,
Configs: customConfigs,
})
if err != nil && !errors.Is(err, ErrUnknownScheme) {
return nil, err
}
}
historyArchiver, err = gcloud.NewHistoryArchiver(p.executionManager, p.logger, p.metricsHandler, p.historyArchiverConfigs.Gstorage)
if historyArchiver == nil {
switch scheme {
case filestore.URIScheme:
if p.historyArchiverConfigs.Filestore == nil {
return nil, ErrArchiverConfigNotFound
}
historyArchiver, err = filestore.NewHistoryArchiver(p.executionManager, p.logger, p.metricsHandler, p.historyArchiverConfigs.Filestore)
case s3store.URIScheme:
if p.historyArchiverConfigs.S3store == nil {
return nil, ErrArchiverConfigNotFound
case gcloud.URIScheme:
if p.historyArchiverConfigs.Gstorage == nil {
return nil, ErrArchiverConfigNotFound
}
historyArchiver, err = gcloud.NewHistoryArchiver(p.executionManager, p.logger, p.metricsHandler, p.historyArchiverConfigs.Gstorage)
case s3store.URIScheme:
if p.historyArchiverConfigs.S3store == nil {
return nil, ErrArchiverConfigNotFound
}
historyArchiver, err = s3store.NewHistoryArchiver(p.executionManager, p.logger, p.metricsHandler, p.historyArchiverConfigs.S3store)
default:
return nil, ErrUnknownScheme
}
historyArchiver, err = s3store.NewHistoryArchiver(p.executionManager, p.logger, p.metricsHandler, p.historyArchiverConfigs.S3store)
default:
return nil, ErrUnknownScheme
}
if err != nil {
@@ -121,25 +194,43 @@ func (p *archiverProvider) GetVisibilityArchiver(scheme string) (archiver.Visibi
var visibilityArchiver archiver.VisibilityArchiver
var err error
switch scheme {
case filestore.URIScheme:
if p.visibilityArchiverConfigs.Filestore == nil {
return nil, ErrArchiverConfigNotFound
if p.customVisibilityArchiverFactory != nil {
var customConfigs map[string]any
if p.visibilityArchiverConfigs != nil {
customConfigs = p.visibilityArchiverConfigs.CustomStores[scheme]
}
visibilityArchiver, err = filestore.NewVisibilityArchiver(p.logger, p.metricsHandler, p.visibilityArchiverConfigs.Filestore)
case s3store.URIScheme:
if p.visibilityArchiverConfigs.S3store == nil {
return nil, ErrArchiverConfigNotFound
visibilityArchiver, err = p.customVisibilityArchiverFactory.NewCustomVisibilityArchiver(NewCustomVisibilityArchiverParams{
Scheme: scheme,
Logger: p.logger,
MetricsHandler: p.metricsHandler,
Configs: customConfigs,
})
if err != nil && !errors.Is(err, ErrUnknownScheme) {
return nil, err
}
visibilityArchiver, err = s3store.NewVisibilityArchiver(p.logger, p.metricsHandler, p.visibilityArchiverConfigs.S3store)
case gcloud.URIScheme:
if p.visibilityArchiverConfigs.Gstorage == nil {
return nil, ErrArchiverConfigNotFound
}
visibilityArchiver, err = gcloud.NewVisibilityArchiver(p.logger, p.metricsHandler, p.visibilityArchiverConfigs.Gstorage)
}
default:
return nil, ErrUnknownScheme
if visibilityArchiver == nil {
switch scheme {
case filestore.URIScheme:
if p.visibilityArchiverConfigs.Filestore == nil {
return nil, ErrArchiverConfigNotFound
}
visibilityArchiver, err = filestore.NewVisibilityArchiver(p.logger, p.metricsHandler, p.visibilityArchiverConfigs.Filestore)
case s3store.URIScheme:
if p.visibilityArchiverConfigs.S3store == nil {
return nil, ErrArchiverConfigNotFound
}
visibilityArchiver, err = s3store.NewVisibilityArchiver(p.logger, p.metricsHandler, p.visibilityArchiverConfigs.S3store)
case gcloud.URIScheme:
if p.visibilityArchiverConfigs.Gstorage == nil {
return nil, ErrArchiverConfigNotFound
}
visibilityArchiver, err = gcloud.NewVisibilityArchiver(p.logger, p.metricsHandler, p.visibilityArchiverConfigs.Gstorage)
default:
return nil, ErrUnknownScheme
}
}
if err != nil {
return nil, err
+78
View File
@@ -69,3 +69,81 @@ func (mr *MockArchiverProviderMockRecorder) GetVisibilityArchiver(scheme any) *g
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetVisibilityArchiver", reflect.TypeOf((*MockArchiverProvider)(nil).GetVisibilityArchiver), scheme)
}
// MockCustomHistoryArchiverFactory is a mock of CustomHistoryArchiverFactory interface.
type MockCustomHistoryArchiverFactory struct {
ctrl *gomock.Controller
recorder *MockCustomHistoryArchiverFactoryMockRecorder
isgomock struct{}
}
// MockCustomHistoryArchiverFactoryMockRecorder is the mock recorder for MockCustomHistoryArchiverFactory.
type MockCustomHistoryArchiverFactoryMockRecorder struct {
mock *MockCustomHistoryArchiverFactory
}
// NewMockCustomHistoryArchiverFactory creates a new mock instance.
func NewMockCustomHistoryArchiverFactory(ctrl *gomock.Controller) *MockCustomHistoryArchiverFactory {
mock := &MockCustomHistoryArchiverFactory{ctrl: ctrl}
mock.recorder = &MockCustomHistoryArchiverFactoryMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockCustomHistoryArchiverFactory) EXPECT() *MockCustomHistoryArchiverFactoryMockRecorder {
return m.recorder
}
// NewCustomHistoryArchiver mocks base method.
func (m *MockCustomHistoryArchiverFactory) NewCustomHistoryArchiver(arg0 NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "NewCustomHistoryArchiver", arg0)
ret0, _ := ret[0].(archiver.HistoryArchiver)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// NewCustomHistoryArchiver indicates an expected call of NewCustomHistoryArchiver.
func (mr *MockCustomHistoryArchiverFactoryMockRecorder) NewCustomHistoryArchiver(arg0 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NewCustomHistoryArchiver", reflect.TypeOf((*MockCustomHistoryArchiverFactory)(nil).NewCustomHistoryArchiver), arg0)
}
// MockCustomVisibilityArchiverFactory is a mock of CustomVisibilityArchiverFactory interface.
type MockCustomVisibilityArchiverFactory struct {
ctrl *gomock.Controller
recorder *MockCustomVisibilityArchiverFactoryMockRecorder
isgomock struct{}
}
// MockCustomVisibilityArchiverFactoryMockRecorder is the mock recorder for MockCustomVisibilityArchiverFactory.
type MockCustomVisibilityArchiverFactoryMockRecorder struct {
mock *MockCustomVisibilityArchiverFactory
}
// NewMockCustomVisibilityArchiverFactory creates a new mock instance.
func NewMockCustomVisibilityArchiverFactory(ctrl *gomock.Controller) *MockCustomVisibilityArchiverFactory {
mock := &MockCustomVisibilityArchiverFactory{ctrl: ctrl}
mock.recorder = &MockCustomVisibilityArchiverFactoryMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockCustomVisibilityArchiverFactory) EXPECT() *MockCustomVisibilityArchiverFactoryMockRecorder {
return m.recorder
}
// NewCustomVisibilityArchiver mocks base method.
func (m *MockCustomVisibilityArchiverFactory) NewCustomVisibilityArchiver(arg0 NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "NewCustomVisibilityArchiver", arg0)
ret0, _ := ret[0].(archiver.VisibilityArchiver)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// NewCustomVisibilityArchiver indicates an expected call of NewCustomVisibilityArchiver.
func (mr *MockCustomVisibilityArchiverFactoryMockRecorder) NewCustomVisibilityArchiver(arg0 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NewCustomVisibilityArchiver", reflect.TypeOf((*MockCustomVisibilityArchiverFactory)(nil).NewCustomVisibilityArchiver), arg0)
}
+618
View File
@@ -0,0 +1,618 @@
package provider
import (
"errors"
"sync"
"testing"
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
"go.temporal.io/server/common/archiver"
"go.temporal.io/server/common/archiver/filestore"
"go.temporal.io/server/common/archiver/gcloud"
"go.temporal.io/server/common/archiver/s3store"
"go.temporal.io/server/common/config"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/persistence"
"go.uber.org/mock/gomock"
)
type (
ProviderSuite struct {
*require.Assertions
suite.Suite
controller *gomock.Controller
mockExecutionManager *persistence.MockExecutionManager
mockHistoryArchiver *archiver.MockHistoryArchiver
mockVisibilityArchiver *archiver.MockVisibilityArchiver
mockCustomHistoryFactory *MockCustomHistoryArchiverFactory
mockCustomVisibilityFactory *MockCustomVisibilityArchiverFactory
logger log.Logger
metricsHandler metrics.Handler
}
)
func TestProviderSuite(t *testing.T) {
suite.Run(t, new(ProviderSuite))
}
func (s *ProviderSuite) SetupTest() {
s.Assertions = require.New(s.T())
s.controller = gomock.NewController(s.T())
s.mockExecutionManager = persistence.NewMockExecutionManager(s.controller)
s.mockHistoryArchiver = archiver.NewMockHistoryArchiver(s.controller)
s.mockVisibilityArchiver = archiver.NewMockVisibilityArchiver(s.controller)
s.mockCustomHistoryFactory = NewMockCustomHistoryArchiverFactory(s.controller)
s.mockCustomVisibilityFactory = NewMockCustomVisibilityArchiverFactory(s.controller)
s.logger = log.NewNoopLogger()
s.metricsHandler = metrics.NoopMetricsHandler
}
func (s *ProviderSuite) TearDownTest() {
s.controller.Finish()
}
func (s *ProviderSuite) TestNewArchiverProvider() {
historyConfig := &config.HistoryArchiverProvider{}
visibilityConfig := &config.VisibilityArchiverProvider{}
provider := NewArchiverProvider(
historyConfig,
visibilityConfig,
s.mockCustomHistoryFactory,
s.mockCustomVisibilityFactory,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
s.NotNil(provider)
// Verify internal state
p := provider.(*archiverProvider)
s.Equal(historyConfig, p.historyArchiverConfigs)
s.Equal(visibilityConfig, p.visibilityArchiverConfigs)
s.Equal(s.mockCustomHistoryFactory, p.customHistoryArchiverFactory)
s.Equal(s.mockCustomVisibilityFactory, p.customVisibilityArchiverFactory)
s.Equal(s.mockExecutionManager, p.executionManager)
s.Equal(s.logger, p.logger)
s.Equal(s.metricsHandler, p.metricsHandler)
s.NotNil(p.historyArchivers)
s.NotNil(p.visibilityArchivers)
}
func (s *ProviderSuite) TestGetHistoryArchiver_CustomFactory_Success() {
scheme := "custom"
provider := NewArchiverProvider(
&config.HistoryArchiverProvider{
CustomStores: map[string]map[string]any{
scheme: {"key": "value"},
},
},
nil,
s.mockCustomHistoryFactory,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
s.mockCustomHistoryFactory.EXPECT().
NewCustomHistoryArchiver(gomock.Any()).
DoAndReturn(func(params NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error) {
s.Equal(scheme, params.Scheme)
s.Equal(s.mockExecutionManager, params.ExecutionManager)
s.Equal(s.logger, params.Logger)
s.Equal(s.metricsHandler, params.MetricsHandler)
s.Equal(map[string]any{"key": "value"}, params.Configs)
return s.mockHistoryArchiver, nil
})
result, err := provider.GetHistoryArchiver(scheme)
s.NoError(err)
s.Equal(s.mockHistoryArchiver, result)
// Test caching - should not call factory again
result2, err := provider.GetHistoryArchiver(scheme)
s.NoError(err)
s.Equal(result, result2)
}
func (s *ProviderSuite) TestGetHistoryArchiver_CustomFactory_ReturnsError() {
scheme := "custom"
expectedErr := errors.New("custom factory error")
provider := NewArchiverProvider(
nil,
nil,
s.mockCustomHistoryFactory,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
s.mockCustomHistoryFactory.EXPECT().
NewCustomHistoryArchiver(gomock.Any()).
Return(nil, expectedErr)
result, err := provider.GetHistoryArchiver(scheme)
s.Error(err)
s.Equal(expectedErr, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetHistoryArchiver_CustomFactory_FallbackToBuiltIn() {
scheme := filestore.URIScheme
provider := NewArchiverProvider(
&config.HistoryArchiverProvider{
Filestore: &config.FilestoreArchiver{
FileMode: "0600",
DirMode: "0700",
},
},
nil,
s.mockCustomHistoryFactory,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
// Custom factory returns ErrUnknownScheme, should fallback to built-in
s.mockCustomHistoryFactory.EXPECT().
NewCustomHistoryArchiver(gomock.Any()).
Return(nil, ErrUnknownScheme)
result, err := provider.GetHistoryArchiver(scheme)
s.NoError(err)
s.NotNil(result)
// Verify it's cached
result2, err := provider.GetHistoryArchiver(scheme)
s.NoError(err)
s.Equal(result, result2)
}
func (s *ProviderSuite) TestGetHistoryArchiver_UnknownScheme() {
scheme := "unknown"
provider := NewArchiverProvider(
nil,
nil,
nil,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
result, err := provider.GetHistoryArchiver(scheme)
s.Error(err)
s.Equal(ErrUnknownScheme, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetHistoryArchiver_ConfigNotFound_Filestore() {
scheme := filestore.URIScheme
provider := NewArchiverProvider(
&config.HistoryArchiverProvider{
Filestore: nil, // Config not set
},
nil,
nil,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
result, err := provider.GetHistoryArchiver(scheme)
s.Error(err)
s.Equal(ErrArchiverConfigNotFound, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetHistoryArchiver_ConfigNotFound_GCloud() {
scheme := gcloud.URIScheme
provider := NewArchiverProvider(
&config.HistoryArchiverProvider{
Gstorage: nil, // Config not set
},
nil,
nil,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
result, err := provider.GetHistoryArchiver(scheme)
s.Error(err)
s.Equal(ErrArchiverConfigNotFound, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetHistoryArchiver_ConfigNotFound_S3() {
scheme := s3store.URIScheme
provider := NewArchiverProvider(
&config.HistoryArchiverProvider{
S3store: nil, // Config not set
},
nil,
nil,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
result, err := provider.GetHistoryArchiver(scheme)
s.Error(err)
s.Equal(ErrArchiverConfigNotFound, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetHistoryArchiver_ConcurrentAccess() {
scheme := "custom"
provider := NewArchiverProvider(
nil,
nil,
s.mockCustomHistoryFactory,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
// The factory may be called multiple times before caching occurs,
// but all calls should return the same archiver.
s.mockCustomHistoryFactory.EXPECT().
NewCustomHistoryArchiver(gomock.Any()).
Return(s.mockHistoryArchiver, nil).
AnyTimes()
var wg sync.WaitGroup
numGoroutines := 10
results := make([]archiver.HistoryArchiver, numGoroutines)
archiverErrors := make([]error, numGoroutines)
for i := range numGoroutines {
wg.Go(func() {
results[i], archiverErrors[i] = provider.GetHistoryArchiver(scheme)
})
}
wg.Wait()
// All should succeed and return a valid archiver
for i := range numGoroutines {
s.NoError(archiverErrors[i])
s.NotNil(results[i])
}
// Verify caching works after concurrent initialization
cachedResult, err := provider.GetHistoryArchiver(scheme)
s.NoError(err)
s.NotNil(cachedResult)
}
func (s *ProviderSuite) TestGetHistoryArchiver_NilConfigs() {
scheme := "custom"
provider := NewArchiverProvider(
nil, // nil history config
nil,
s.mockCustomHistoryFactory,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
s.mockCustomHistoryFactory.EXPECT().
NewCustomHistoryArchiver(gomock.Any()).
DoAndReturn(func(params NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error) {
s.Nil(params.Configs) // Should be nil when configs are not provided
return s.mockHistoryArchiver, nil
})
result, err := provider.GetHistoryArchiver(scheme)
s.NoError(err)
s.Equal(s.mockHistoryArchiver, result)
}
func (s *ProviderSuite) TestCustomHistoryArchiverFactoryFunc() {
called := false
expectedArchiver := s.mockHistoryArchiver
factoryFunc := CustomHistoryArchiverFactoryFunc(func(params NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error) {
called = true
s.Equal("test-scheme", params.Scheme)
return expectedArchiver, nil
})
result, err := factoryFunc.NewCustomHistoryArchiver(NewCustomHistoryArchiverParams{
Scheme: "test-scheme",
})
s.NoError(err)
s.True(called)
s.Equal(expectedArchiver, result)
}
func (s *ProviderSuite) TestGetVisibilityArchiver_CustomFactory_Success() {
scheme := "custom"
provider := NewArchiverProvider(
nil,
&config.VisibilityArchiverProvider{
CustomStores: map[string]map[string]any{
scheme: {"key": "value"},
},
},
nil,
s.mockCustomVisibilityFactory,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
s.mockCustomVisibilityFactory.EXPECT().
NewCustomVisibilityArchiver(gomock.Any()).
DoAndReturn(func(params NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error) {
s.Equal(scheme, params.Scheme)
s.Equal(s.logger, params.Logger)
s.Equal(s.metricsHandler, params.MetricsHandler)
s.Equal(map[string]any{"key": "value"}, params.Configs)
return s.mockVisibilityArchiver, nil
})
result, err := provider.GetVisibilityArchiver(scheme)
s.NoError(err)
s.Equal(s.mockVisibilityArchiver, result)
// Test caching - should not call factory again
result2, err := provider.GetVisibilityArchiver(scheme)
s.NoError(err)
s.Equal(result, result2)
}
func (s *ProviderSuite) TestGetVisibilityArchiver_CustomFactory_ReturnsError() {
scheme := "custom"
expectedErr := errors.New("custom factory error")
provider := NewArchiverProvider(
nil,
nil,
nil,
s.mockCustomVisibilityFactory,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
s.mockCustomVisibilityFactory.EXPECT().
NewCustomVisibilityArchiver(gomock.Any()).
Return(nil, expectedErr)
result, err := provider.GetVisibilityArchiver(scheme)
s.Error(err)
s.Equal(expectedErr, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetVisibilityArchiver_CustomFactory_FallbackToBuiltIn() {
scheme := filestore.URIScheme
provider := NewArchiverProvider(
nil,
&config.VisibilityArchiverProvider{
Filestore: &config.FilestoreArchiver{
FileMode: "0600",
DirMode: "0700",
},
},
nil,
s.mockCustomVisibilityFactory,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
// Custom factory returns ErrUnknownScheme, should fallback to built-in
s.mockCustomVisibilityFactory.EXPECT().
NewCustomVisibilityArchiver(gomock.Any()).
Return(nil, ErrUnknownScheme)
result, err := provider.GetVisibilityArchiver(scheme)
s.NoError(err)
s.NotNil(result)
// Verify it's cached
result2, err := provider.GetVisibilityArchiver(scheme)
s.NoError(err)
s.Equal(result, result2)
}
func (s *ProviderSuite) TestGetVisibilityArchiver_UnknownScheme() {
scheme := "unknown"
provider := NewArchiverProvider(
nil,
nil,
nil,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
result, err := provider.GetVisibilityArchiver(scheme)
s.Error(err)
s.Equal(ErrUnknownScheme, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetVisibilityArchiver_ConfigNotFound_Filestore() {
scheme := filestore.URIScheme
provider := NewArchiverProvider(
nil,
&config.VisibilityArchiverProvider{
Filestore: nil, // Config not set
},
nil,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
result, err := provider.GetVisibilityArchiver(scheme)
s.Error(err)
s.Equal(ErrArchiverConfigNotFound, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetVisibilityArchiver_ConfigNotFound_GCloud() {
scheme := gcloud.URIScheme
provider := NewArchiverProvider(
nil,
&config.VisibilityArchiverProvider{
Gstorage: nil, // Config not set
},
nil,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
result, err := provider.GetVisibilityArchiver(scheme)
s.Error(err)
s.Equal(ErrArchiverConfigNotFound, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetVisibilityArchiver_ConfigNotFound_S3() {
scheme := s3store.URIScheme
provider := NewArchiverProvider(
nil,
&config.VisibilityArchiverProvider{
S3store: nil, // Config not set
},
nil,
nil,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
result, err := provider.GetVisibilityArchiver(scheme)
s.Error(err)
s.Equal(ErrArchiverConfigNotFound, err)
s.Nil(result)
}
func (s *ProviderSuite) TestGetVisibilityArchiver_ConcurrentAccess() {
scheme := "custom"
provider := NewArchiverProvider(
nil,
nil,
nil,
s.mockCustomVisibilityFactory,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
// The factory may be called multiple times before caching occurs,
// but all calls should return the same archiver.
s.mockCustomVisibilityFactory.EXPECT().
NewCustomVisibilityArchiver(gomock.Any()).
Return(s.mockVisibilityArchiver, nil).
AnyTimes()
var wg sync.WaitGroup
numGoroutines := 10
results := make([]archiver.VisibilityArchiver, numGoroutines)
archiverErrors := make([]error, numGoroutines)
for i := range numGoroutines {
wg.Go(func() {
results[i], archiverErrors[i] = provider.GetVisibilityArchiver(scheme)
})
}
wg.Wait()
// All should succeed and return a valid archiver
for i := range numGoroutines {
s.NoError(archiverErrors[i])
s.NotNil(results[i])
}
// Verify caching works after concurrent initialization
cachedResult, err := provider.GetVisibilityArchiver(scheme)
s.NoError(err)
s.NotNil(cachedResult)
}
func (s *ProviderSuite) TestGetVisibilityArchiver_NilConfigs() {
scheme := "custom"
provider := NewArchiverProvider(
nil,
nil, // nil visibility config
nil,
s.mockCustomVisibilityFactory,
s.mockExecutionManager,
s.logger,
s.metricsHandler,
)
s.mockCustomVisibilityFactory.EXPECT().
NewCustomVisibilityArchiver(gomock.Any()).
DoAndReturn(func(params NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error) {
s.Nil(params.Configs) // Should be nil when configs are not provided
return s.mockVisibilityArchiver, nil
})
result, err := provider.GetVisibilityArchiver(scheme)
s.NoError(err)
s.Equal(s.mockVisibilityArchiver, result)
}
func (s *ProviderSuite) TestCustomVisibilityArchiverFactoryFunc() {
called := false
expectedArchiver := s.mockVisibilityArchiver
factoryFunc := CustomVisibilityArchiverFactoryFunc(func(params NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error) {
called = true
s.Equal("test-scheme", params.Scheme)
return expectedArchiver, nil
})
result, err := factoryFunc.NewCustomVisibilityArchiver(NewCustomVisibilityArchiverParams{
Scheme: "test-scheme",
})
s.NoError(err)
s.True(called)
s.Equal(expectedArchiver, result)
}
+6
View File
@@ -484,6 +484,9 @@ type (
Filestore *FilestoreArchiver `yaml:"filestore"`
Gstorage *GstorageArchiver `yaml:"gstorage"`
S3store *S3Archiver `yaml:"s3store"`
// CustomStores contains the config for all custom history archivers
// The structure is a map of archiver name (scheme) to a map of config key-values
CustomStores map[string]map[string]any `yaml:"customStores"`
}
// VisibilityArchival contains the config for visibility archival
@@ -501,6 +504,9 @@ type (
Filestore *FilestoreArchiver `yaml:"filestore"`
S3store *S3Archiver `yaml:"s3store"`
Gstorage *GstorageArchiver `yaml:"gstorage"`
// CustomStores contains the config for all custom visibility archivers
// The structure is a map of archiver name (scheme) to a map of config key-values
CustomStores map[string]map[string]any `yaml:"customStores"`
}
// FilestoreArchiver contain the config for filestore archiver
+4
View File
@@ -344,6 +344,8 @@ func ArchivalMetadataProvider(dc *dynamicconfig.Collection, cfg *config.Config)
func ArchiverProviderProvider(
cfg *config.Config,
customHistoryArchiverFactory provider.CustomHistoryArchiverFactory,
customVisibilityArchiverFactory provider.CustomVisibilityArchiverFactory,
persistenceExecutionManager persistence.ExecutionManager,
logger log.SnTaggedLogger,
metricsHandler metrics.Handler,
@@ -351,6 +353,8 @@ func ArchiverProviderProvider(
return provider.NewArchiverProvider(
cfg.Archival.History.Provider,
cfg.Archival.Visibility.Provider,
customHistoryArchiverFactory,
customVisibilityArchiverFactory,
persistenceExecutionManager,
logger,
metricsHandler,
+43 -30
View File
@@ -26,6 +26,7 @@ import (
chasmworkflow "go.temporal.io/server/chasm/lib/workflow"
"go.temporal.io/server/client"
"go.temporal.io/server/common/archiver"
"go.temporal.io/server/common/archiver/provider"
"go.temporal.io/server/common/authorization"
"go.temporal.io/server/common/cluster"
"go.temporal.io/server/common/config"
@@ -105,9 +106,11 @@ type (
ServiceNames resource.ServiceNames
NamespaceLogger resource.NamespaceLogger
ServiceResolver resolver.ServiceResolver
CustomDataStoreFactory persistenceClient.AbstractDataStoreFactory
CustomVisibilityStore visibility.VisibilityStoreFactory
ServiceResolver resolver.ServiceResolver
CustomDataStoreFactory persistenceClient.AbstractDataStoreFactory
CustomVisibilityStore visibility.VisibilityStoreFactory
CustomHistoryArchiverFactory provider.CustomHistoryArchiverFactory
CustomVisibilityArchiverFactory provider.CustomVisibilityArchiverFactory
SearchAttributesMapper searchattribute.Mapper
CustomFrontendInterceptors []grpc.UnaryServerInterceptor
@@ -289,9 +292,11 @@ func ServerOptionsProvider(opts []ServerOption) (serverOptionsProvider, error) {
ServiceHosts: so.hostsByService,
NamespaceLogger: so.namespaceLogger,
ServiceResolver: so.persistenceServiceResolver,
CustomDataStoreFactory: so.customDataStoreFactory,
CustomVisibilityStore: so.customVisibilityStoreFactory,
ServiceResolver: so.persistenceServiceResolver,
CustomDataStoreFactory: so.customDataStoreFactory,
CustomVisibilityStore: so.customVisibilityStoreFactory,
CustomHistoryArchiverFactory: so.customHistoryArchiverFactory,
CustomVisibilityArchiverFactory: so.customVisibilityArchiverFactory,
SearchAttributesMapper: so.searchAttributesMapper,
CustomFrontendInterceptors: so.customFrontendInterceptors,
@@ -344,30 +349,32 @@ type (
ServiceProviderParamsCommon struct {
fx.In
Cfg *config.Config
ServiceNames resource.ServiceNames
Logger log.Logger
NamespaceLogger resource.NamespaceLogger
DynamicConfigClient dynamicconfig.Client
MetricsHandler metrics.Handler
EsClient esclient.Client
TlsConfigProvider encryption.TLSConfigProvider
PersistenceConfig config.Persistence
ClusterMetadata *cluster.Config
ClientFactoryProvider client.FactoryProvider
AudienceGetter authorization.JWTAudienceMapper
PersistenceServiceResolver resolver.ServiceResolver
PersistenceFactoryProvider persistenceClient.FactoryProviderFn
SearchAttributesMapper searchattribute.Mapper
CustomFrontendInterceptors []grpc.UnaryServerInterceptor
Authorizer authorization.Authorizer
ClaimMapper authorization.ClaimMapper
DataStoreFactory persistenceClient.AbstractDataStoreFactory
VisibilityStoreFactory visibility.VisibilityStoreFactory
SpanExporters []otelsdktrace.SpanExporter
InstanceID resource.InstanceID `optional:"true"`
StaticServiceHosts map[primitives.ServiceName]static.Hosts `optional:"true"`
TaskCategoryRegistry tasks.TaskCategoryRegistry
Cfg *config.Config
ServiceNames resource.ServiceNames
Logger log.Logger
NamespaceLogger resource.NamespaceLogger
DynamicConfigClient dynamicconfig.Client
MetricsHandler metrics.Handler
EsClient esclient.Client
TlsConfigProvider encryption.TLSConfigProvider //nolint:staticcheck // should be TLSConfigProvider
PersistenceConfig config.Persistence
ClusterMetadata *cluster.Config
ClientFactoryProvider client.FactoryProvider
AudienceGetter authorization.JWTAudienceMapper
PersistenceServiceResolver resolver.ServiceResolver
PersistenceFactoryProvider persistenceClient.FactoryProviderFn
SearchAttributesMapper searchattribute.Mapper
CustomFrontendInterceptors []grpc.UnaryServerInterceptor
Authorizer authorization.Authorizer
ClaimMapper authorization.ClaimMapper
DataStoreFactory persistenceClient.AbstractDataStoreFactory
VisibilityStoreFactory visibility.VisibilityStoreFactory
CustomHistoryArchiverFactory provider.CustomHistoryArchiverFactory
CustomVisibilityArchiverFactory provider.CustomVisibilityArchiverFactory
SpanExporters []otelsdktrace.SpanExporter
InstanceID resource.InstanceID `optional:"true"`
StaticServiceHosts map[primitives.ServiceName]static.Hosts `optional:"true"`
TaskCategoryRegistry tasks.TaskCategoryRegistry
}
)
@@ -401,6 +408,12 @@ func (params ServiceProviderParamsCommon) GetCommonServiceOptions(serviceName pr
func() visibility.VisibilityStoreFactory {
return params.VisibilityStoreFactory
},
func() provider.CustomHistoryArchiverFactory {
return params.CustomHistoryArchiverFactory
},
func() provider.CustomVisibilityArchiverFactory {
return params.CustomVisibilityArchiverFactory
},
func() client.FactoryProvider {
return params.ClientFactoryProvider
},
+17
View File
@@ -4,6 +4,7 @@ import (
"net/http"
"go.temporal.io/server/client"
"go.temporal.io/server/common/archiver/provider"
"go.temporal.io/server/common/authorization"
"go.temporal.io/server/common/config"
"go.temporal.io/server/common/dynamicconfig"
@@ -154,6 +155,22 @@ func WithCustomVisibilityStoreFactory(customFactory visibility.VisibilityStoreFa
})
}
// WithCustomHistoryArchiverFactory sets a custom history archiver factory.
// NOTE: this option is experimental and may be changed in future release.
func WithCustomHistoryArchiverFactory(factory provider.CustomHistoryArchiverFactory) ServerOption {
return applyFunc(func(s *serverOptions) {
s.customHistoryArchiverFactory = factory
})
}
// WithCustomVisibilityArchiverFactory sets a custom visibility archiver factory.
// NOTE: this option is experimental and may be changed in future release.
func WithCustomVisibilityArchiverFactory(factory provider.CustomVisibilityArchiverFactory) ServerOption {
return applyFunc(func(s *serverOptions) {
s.customVisibilityArchiverFactory = factory
})
}
// WithClientFactoryProvider sets a custom ClientFactoryProvider
// NOTE: this option is experimental and may be changed or removed in future release.
func WithClientFactoryProvider(clientFactoryProvider client.FactoryProvider) ServerOption {
+18 -15
View File
@@ -7,6 +7,7 @@ import (
"slices"
"go.temporal.io/server/client"
"go.temporal.io/server/common/archiver/provider"
"go.temporal.io/server/common/authorization"
"go.temporal.io/server/common/config"
"go.temporal.io/server/common/dynamicconfig"
@@ -40,21 +41,23 @@ type (
startupSynchronizationMode synchronizationModeParams
logger log.Logger
namespaceLogger log.Logger
authorizer authorization.Authorizer
tlsConfigProvider encryption.TLSConfigProvider
claimMapper authorization.ClaimMapper
audienceGetter authorization.JWTAudienceMapper
persistenceServiceResolver resolver.ServiceResolver
elasticsearchHttpClient *http.Client
dynamicConfigClient dynamicconfig.Client
customDataStoreFactory persistenceClient.AbstractDataStoreFactory
customVisibilityStoreFactory visibility.VisibilityStoreFactory
clientFactoryProvider client.FactoryProvider
searchAttributesMapper searchattribute.Mapper
customFrontendInterceptors []grpc.UnaryServerInterceptor
metricHandler metrics.Handler
logger log.Logger
namespaceLogger log.Logger
authorizer authorization.Authorizer
tlsConfigProvider encryption.TLSConfigProvider
claimMapper authorization.ClaimMapper
audienceGetter authorization.JWTAudienceMapper
persistenceServiceResolver resolver.ServiceResolver
elasticsearchHttpClient *http.Client //nolint:staticcheck // should be elasticsearchHTTPClient
dynamicConfigClient dynamicconfig.Client
customDataStoreFactory persistenceClient.AbstractDataStoreFactory
customVisibilityStoreFactory visibility.VisibilityStoreFactory
customHistoryArchiverFactory provider.CustomHistoryArchiverFactory
customVisibilityArchiverFactory provider.CustomVisibilityArchiverFactory
clientFactoryProvider client.FactoryProvider
searchAttributesMapper searchattribute.Mapper
customFrontendInterceptors []grpc.UnaryServerInterceptor
metricHandler metrics.Handler
}
)
+128
View File
@@ -2,9 +2,11 @@ package tests
import (
"bytes"
"context"
"encoding/binary"
"fmt"
"strconv"
"sync/atomic"
"testing"
"time"
@@ -17,9 +19,11 @@ import (
workflowpb "go.temporal.io/api/workflow/v1"
"go.temporal.io/api/workflowservice/v1"
"go.temporal.io/server/api/adminservice/v1"
archiverspb "go.temporal.io/server/api/archiver/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/common"
"go.temporal.io/server/common/archiver"
"go.temporal.io/server/common/archiver/provider"
"go.temporal.io/server/common/convert"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/log/tag"
@@ -33,20 +37,71 @@ import (
"google.golang.org/protobuf/types/known/durationpb"
)
const (
// Custom scheme for testing custom archiver implementation
customArchiverScheme = "customtest"
)
type (
ArchivalSuite struct {
testcore.FunctionalTestBase
archivalNamespace namespace.Name
archivalNamespaceID namespace.ID
// Namespace for testing custom archiver
customArchiverNamespace namespace.Name
customArchiverNamespaceID namespace.ID
// Counters to verify custom archivers are being called
customHistoryArchiveCalled atomic.Int32
customVisibilityArchiveCalled atomic.Int32
}
archivalWorkflowInfo struct {
execution *commonpb.WorkflowExecution
branchToken []byte
}
// customHistoryArchiver wraps a built-in history archiver and tracks Archive calls
customHistoryArchiver struct {
counter *atomic.Int32
}
// customVisibilityArchiver wraps a built-in visibility archiver and tracks Archive calls
customVisibilityArchiver struct {
counter *atomic.Int32
}
)
// customHistoryArchiver method implementations
func (c *customHistoryArchiver) Archive(ctx context.Context, uri archiver.URI, request *archiver.ArchiveHistoryRequest, opts ...archiver.ArchiveOption) error {
c.counter.Add(1)
return nil
}
func (c *customHistoryArchiver) Get(ctx context.Context, uri archiver.URI, request *archiver.GetHistoryRequest) (*archiver.GetHistoryResponse, error) {
return nil, nil
}
func (c *customHistoryArchiver) ValidateURI(uri archiver.URI) error {
return nil
}
// customVisibilityArchiver method implementations
func (c *customVisibilityArchiver) Archive(ctx context.Context, uri archiver.URI, request *archiverspb.VisibilityRecord, opts ...archiver.ArchiveOption) error {
c.counter.Add(1)
return nil
}
func (c *customVisibilityArchiver) Query(ctx context.Context, uri archiver.URI, request *archiver.QueryVisibilityRequest, saTypeMap searchattribute.NameTypeMap) (*archiver.QueryVisibilityResponse, error) {
return nil, nil
}
func (c *customVisibilityArchiver) ValidateURI(uri archiver.URI) error {
return nil
}
func TestArchivalSuite(t *testing.T) {
t.Parallel() // This suite can work in parallel as long as it is the only one that use testcore.WithArchivalEnabled() option.
suite.Run(t, new(ArchivalSuite))
@@ -57,12 +112,44 @@ func (s *ArchivalSuite) SetupSuite() {
dynamicconfig.ArchivalProcessorArchiveDelay.Key(): time.Duration(0),
}
// Create custom history archiver factory for custom scheme
customHistoryArchiverFactory := provider.CustomHistoryArchiverFactoryFunc(
func(params provider.NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error) {
// Only handle custom scheme, return ErrUnknownScheme for others (including filestore)
if params.Scheme != customArchiverScheme {
return nil, provider.ErrUnknownScheme
}
// Return a wrapper that delegates to filestore but tracks Archive calls
return &customHistoryArchiver{
counter: &s.customHistoryArchiveCalled,
}, nil
},
)
// Create custom visibility archiver factory for custom scheme
customVisibilityArchiverFactory := provider.CustomVisibilityArchiverFactoryFunc(
func(params provider.NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error) {
// Only handle custom scheme, return ErrUnknownScheme for others (including filestore)
if params.Scheme != customArchiverScheme {
return nil, provider.ErrUnknownScheme
}
// Return a wrapper that delegates to filestore but tracks Archive calls
return &customVisibilityArchiver{
counter: &s.customVisibilityArchiveCalled,
}, nil
},
)
s.FunctionalTestBase.SetupSuiteWithCluster(
testcore.WithDynamicConfigOverrides(dynamicConfigOverrides),
testcore.WithArchivalEnabled(),
testcore.WithCustomHistoryArchiverFactory(customHistoryArchiverFactory),
testcore.WithCustomVisibilityArchiverFactory(customVisibilityArchiverFactory),
)
var err error
// Register namespace using built-in filestore archiver
s.archivalNamespace = namespace.Name(testcore.RandomizeStr("archival-enabled-namespace"))
s.archivalNamespaceID, err = s.RegisterNamespace(
s.archivalNamespace,
@@ -72,10 +159,24 @@ func (s *ArchivalSuite) SetupSuite() {
s.GetTestCluster().ArchiverBase().VisibilityURI(),
)
s.Require().NoError(err)
// Register namespace using custom archiver with custom scheme
s.customArchiverNamespace = namespace.Name(testcore.RandomizeStr("custom-archiver-namespace"))
customHistoryURI := customArchiverScheme + "://custom-history-archiver"
customVisibilityURI := customArchiverScheme + "://custom-visibility-archiver"
s.customArchiverNamespaceID, err = s.RegisterNamespace(
s.customArchiverNamespace,
0, // Archive right away.
enumspb.ARCHIVAL_STATE_ENABLED,
customHistoryURI,
customVisibilityURI,
)
s.Require().NoError(err)
}
func (s *ArchivalSuite) TearDownSuite() {
s.Require().NoError(s.MarkNamespaceAsDeleted(s.archivalNamespace))
s.Require().NoError(s.MarkNamespaceAsDeleted(s.customArchiverNamespace))
s.FunctionalTestBase.TearDownCluster()
}
@@ -175,6 +276,33 @@ func (s *ArchivalSuite) TestVisibilityArchival() {
}
}
func (s *ArchivalSuite) TestCustomArchiver() {
s.True(s.GetTestCluster().ArchiverBase().Metadata().GetHistoryConfig().ClusterConfiguredForArchival())
workflowID := "custom-history-archiver-workflow-id"
workflowType := "custom-history-archiver-type"
taskQueue := "custom-history-archiver-task-queue"
numActivities := 1
numRuns := 1
// Reset counter before test
s.customHistoryArchiveCalled.Store(0)
s.customVisibilityArchiveCalled.Store(0)
// Use custom archiver namespace to trigger custom archiver
s.startAndFinishWorkflow(workflowID, workflowType, taskQueue, s.customArchiverNamespace, numActivities, numRuns)
// Verify custom archiver's Archive method was called at least once
s.Eventually(func() bool {
called := s.customHistoryArchiveCalled.Load()
return called > 0
}, 10*time.Second, 500*time.Millisecond, "Custom history archiver Archive method should have been called")
s.Eventually(func() bool {
called := s.customVisibilityArchiveCalled.Load()
return called > 0
}, 10*time.Second, 500*time.Millisecond, "Custom visibility archiver Archive method should have been called")
}
// workflowIsArchived asserts that both the workflow history and workflow visibility are archived.
func (s *ArchivalSuite) workflowIsArchived(namespaceID namespace.ID, execution *commonpb.WorkflowExecution) {
historyURI, err := archiver.NewURI(s.GetTestCluster().ArchiverBase().HistoryURI())
+29 -12
View File
@@ -26,6 +26,7 @@ import (
"go.temporal.io/server/api/adminservice/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
"go.temporal.io/server/common"
"go.temporal.io/server/common/archiver/provider"
"go.temporal.io/server/common/config"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/log"
@@ -92,13 +93,15 @@ type (
}
// TestClusterParams contains the variables which are used to configure test cluster via the TestClusterOption type.
TestClusterParams struct {
ServiceOptions map[primitives.ServiceName][]fx.Option
DynamicConfigOverrides map[dynamicconfig.Key]any
ArchivalEnabled bool
EnableMTLS bool
FaultInjectionConfig *config.FaultInjection
NumHistoryShards int32
SharedCluster bool
ServiceOptions map[primitives.ServiceName][]fx.Option
DynamicConfigOverrides map[dynamicconfig.Key]any
ArchivalEnabled bool
EnableMTLS bool
FaultInjectionConfig *config.FaultInjection
NumHistoryShards int32
SharedCluster bool
CustomHistoryArchiverFactory provider.CustomHistoryArchiverFactory
CustomVisibilityArchiverFactory provider.CustomVisibilityArchiverFactory
}
TestClusterOption func(params *TestClusterParams)
)
@@ -167,6 +170,18 @@ func WithSharedCluster() TestClusterOption {
}
}
func WithCustomHistoryArchiverFactory(factory provider.CustomHistoryArchiverFactory) TestClusterOption {
return func(params *TestClusterParams) {
params.CustomHistoryArchiverFactory = factory
}
}
func WithCustomVisibilityArchiverFactory(factory provider.CustomVisibilityArchiverFactory) TestClusterOption {
return func(params *TestClusterParams) {
params.CustomVisibilityArchiverFactory = factory
}
}
func (s *FunctionalTestBase) GetTestCluster() *TestCluster {
return s.testCluster
}
@@ -268,11 +283,13 @@ func (s *FunctionalTestBase) setupCluster(options ...TestClusterOption) {
HistoryConfig: HistoryConfig{
NumHistoryShards: cmp.Or(params.NumHistoryShards, 4),
},
DynamicConfigOverrides: params.DynamicConfigOverrides,
ServiceFxOptions: params.ServiceOptions,
EnableMetricsCapture: true,
EnableArchival: params.ArchivalEnabled,
EnableMTLS: params.EnableMTLS,
DynamicConfigOverrides: params.DynamicConfigOverrides,
ServiceFxOptions: params.ServiceOptions,
EnableMetricsCapture: true,
EnableArchival: params.ArchivalEnabled,
EnableMTLS: params.EnableMTLS,
CustomHistoryArchiverFactory: params.CustomHistoryArchiverFactory,
CustomVisibilityArchiverFactory: params.CustomVisibilityArchiverFactory,
}
// Apply configuration for shared clusters.
+40 -19
View File
@@ -72,21 +72,23 @@ type (
// TestClusterConfig are config for a test cluster
TestClusterConfig struct {
EnableArchival bool
IsMasterCluster bool
ClusterMetadata cluster.Config
Persistence persistencetests.TestBaseOptions
FrontendConfig FrontendConfig
HistoryConfig HistoryConfig
MatchingConfig MatchingConfig
WorkerConfig WorkerConfig
ESConfig *esclient.Config
MockAdminClient map[string]adminservice.AdminServiceClient
FaultInjection *config.FaultInjection
DynamicConfigOverrides map[dynamicconfig.Key]any
EnableMTLS bool
EnableMetricsCapture bool
SpanExporters map[telemetry.SpanExporterType]sdktrace.SpanExporter
EnableArchival bool
IsMasterCluster bool
ClusterMetadata cluster.Config
Persistence persistencetests.TestBaseOptions
FrontendConfig FrontendConfig
HistoryConfig HistoryConfig
MatchingConfig MatchingConfig
WorkerConfig WorkerConfig
ESConfig *esclient.Config
MockAdminClient map[string]adminservice.AdminServiceClient
FaultInjection *config.FaultInjection
DynamicConfigOverrides map[dynamicconfig.Key]any
EnableMTLS bool
EnableMetricsCapture bool
SpanExporters map[telemetry.SpanExporterType]sdktrace.SpanExporter
CustomHistoryArchiverFactory provider.CustomHistoryArchiverFactory
CustomVisibilityArchiverFactory provider.CustomVisibilityArchiverFactory
// ServiceFxOptions can be populated using WithFxOptionsForService.
ServiceFxOptions map[primitives.ServiceName][]fx.Option
}
@@ -162,7 +164,12 @@ func (f *defaultPersistenceTestBaseFactory) NewTestBase(options *persistencetest
return persistencetests.NewTestBase(options)
}
func newClusterWithPersistenceTestBaseFactory(t *testing.T, clusterConfig *TestClusterConfig, logger log.Logger, tbFactory PersistenceTestBaseFactory) (*TestCluster, error) {
func newClusterWithPersistenceTestBaseFactory(
t *testing.T,
clusterConfig *TestClusterConfig,
logger log.Logger,
tbFactory PersistenceTestBaseFactory,
) (*TestCluster, error) {
// determine number of hosts per service
const minNodes = 1
clusterConfig.FrontendConfig.NumFrontendHosts = max(minNodes, clusterConfig.FrontendConfig.NumFrontendHosts)
@@ -213,7 +220,13 @@ func newClusterWithPersistenceTestBaseFactory(t *testing.T, clusterConfig *TestC
testBase := tbFactory.NewTestBase(&clusterConfig.Persistence)
testBase.Setup(clusterMetadataConfig)
archiverBase := newArchiverBase(clusterConfig.EnableArchival, testBase.ExecutionManager, logger)
archiverBase := newArchiverBase(
clusterConfig.EnableArchival,
clusterConfig.CustomHistoryArchiverFactory,
clusterConfig.CustomVisibilityArchiverFactory,
testBase.ExecutionManager,
logger,
)
pConfig := testBase.DefaultTestCluster.Config()
pConfig.NumHistoryShards = clusterConfig.HistoryConfig.NumHistoryShards
@@ -465,12 +478,18 @@ func newPProfInitializerImpl(logger log.Logger, port int) *pprof.PProfInitialize
}
}
func newArchiverBase(enabled bool, executionManager persistence.ExecutionManager, logger log.Logger) *ArchiverBase {
func newArchiverBase(
enabled bool,
customHistoryArchiverFactory provider.CustomHistoryArchiverFactory,
customVisibilityArchiverFactory provider.CustomVisibilityArchiverFactory,
executionManager persistence.ExecutionManager,
logger log.Logger,
) *ArchiverBase {
dcCollection := dynamicconfig.NewNoopCollection()
if !enabled {
return &ArchiverBase{
metadata: archiver.NewArchivalMetadata(dcCollection, "", false, "", false, &config.ArchivalNamespaceDefaults{}),
provider: provider.NewArchiverProvider(nil, nil, nil, logger, metrics.NoopMetricsHandler),
provider: provider.NewArchiverProvider(nil, nil, nil, nil, nil, logger, metrics.NoopMetricsHandler),
}
}
@@ -493,6 +512,8 @@ func newArchiverBase(enabled bool, executionManager persistence.ExecutionManager
&config.VisibilityArchiverProvider{
Filestore: cfg,
},
customHistoryArchiverFactory,
customVisibilityArchiverFactory,
executionManager,
logger,
metrics.NoopMetricsHandler,