Files
tasks/common/archiver/archivalMetadata.go
2023-01-24 14:51:33 -08:00

237 lines
8.0 KiB
Go

// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
//go:generate mockgen -copyright_file ../../LICENSE -package $GOPACKAGE -source $GOFILE -destination archivalMetadata_mock.go
package archiver
import (
"fmt"
"strings"
enumspb "go.temporal.io/api/enums/v1"
"go.temporal.io/server/common/config"
"go.temporal.io/server/common/dynamicconfig"
)
type (
// ArchivalMetadata provides cluster level archival information
ArchivalMetadata interface {
GetHistoryConfig() ArchivalConfig
GetVisibilityConfig() ArchivalConfig
}
// ArchivalConfig is an immutable representation of the archival configuration of the cluster
// This config is determined at cluster startup time
ArchivalConfig interface {
ClusterConfiguredForArchival() bool
GetClusterState() ArchivalState
ReadEnabled() bool
GetNamespaceDefaultState() enumspb.ArchivalState
GetNamespaceDefaultURI() string
StaticClusterState() ArchivalState
}
archivalMetadata struct {
historyConfig ArchivalConfig
visibilityConfig ArchivalConfig
}
archivalConfig struct {
staticClusterState ArchivalState
dynamicClusterState dynamicconfig.StringPropertyFn
enableRead dynamicconfig.BoolPropertyFn
namespaceDefaultState enumspb.ArchivalState
namespaceDefaultURI string
}
// ArchivalState represents the archival state of the cluster
ArchivalState int
)
func (a *archivalConfig) StaticClusterState() ArchivalState {
return a.staticClusterState
}
const (
// ArchivalDisabled means this cluster is not configured to handle archival
ArchivalDisabled ArchivalState = iota
// ArchivalPaused means this cluster is configured to handle archival but is currently not archiving
// This state is not yet implemented, as of now ArchivalPaused is treated the same way as ArchivalDisabled
ArchivalPaused
// ArchivalEnabled means this cluster is currently archiving
ArchivalEnabled
)
// NewArchivalMetadata constructs a new ArchivalMetadata
func NewArchivalMetadata(
dc *dynamicconfig.Collection,
historyState string,
historyReadEnabled bool,
visibilityState string,
visibilityReadEnabled bool,
namespaceDefaults *config.ArchivalNamespaceDefaults,
) ArchivalMetadata {
historyConfig := NewArchivalConfig(
historyState,
dc.GetStringProperty(dynamicconfig.HistoryArchivalState, historyState),
dc.GetBoolProperty(dynamicconfig.EnableReadFromHistoryArchival, historyReadEnabled),
namespaceDefaults.History.State,
namespaceDefaults.History.URI,
)
visibilityConfig := NewArchivalConfig(
visibilityState,
dc.GetStringProperty(dynamicconfig.VisibilityArchivalState, visibilityState),
dc.GetBoolProperty(dynamicconfig.EnableReadFromVisibilityArchival, visibilityReadEnabled),
namespaceDefaults.Visibility.State,
namespaceDefaults.Visibility.URI,
)
return &archivalMetadata{
historyConfig: historyConfig,
visibilityConfig: visibilityConfig,
}
}
func (metadata *archivalMetadata) GetHistoryConfig() ArchivalConfig {
return metadata.historyConfig
}
func (metadata *archivalMetadata) GetVisibilityConfig() ArchivalConfig {
return metadata.visibilityConfig
}
// NewArchivalConfig constructs a new valid ArchivalConfig
func NewArchivalConfig(
staticClusterStateStr string,
dynamicClusterState dynamicconfig.StringPropertyFn,
enableRead dynamicconfig.BoolPropertyFn,
namespaceDefaultStateStr string,
namespaceDefaultURI string,
) ArchivalConfig {
staticClusterState, err := getClusterArchivalState(staticClusterStateStr)
if err != nil {
panic(err)
}
namespaceDefaultState, err := getNamespaceArchivalState(namespaceDefaultStateStr)
if err != nil {
panic(err)
}
return &archivalConfig{
staticClusterState: staticClusterState,
dynamicClusterState: dynamicClusterState,
enableRead: enableRead,
namespaceDefaultState: namespaceDefaultState,
namespaceDefaultURI: namespaceDefaultURI,
}
}
// NewDisabledArchvialConfig returns an ArchivalConfig where archival is disabled for both the cluster and the namespace
func NewDisabledArchvialConfig() ArchivalConfig {
return &archivalConfig{
staticClusterState: ArchivalDisabled,
dynamicClusterState: nil,
enableRead: nil,
namespaceDefaultState: enumspb.ARCHIVAL_STATE_DISABLED,
namespaceDefaultURI: "",
}
}
// NewEnabledArchivalConfig returns an ArchivalConfig where archival is enabled for both the cluster and the namespace
func NewEnabledArchivalConfig() ArchivalConfig {
return &archivalConfig{
staticClusterState: ArchivalEnabled,
dynamicClusterState: dynamicconfig.GetStringPropertyFn("enabled"),
enableRead: dynamicconfig.GetBoolPropertyFn(true),
namespaceDefaultState: enumspb.ARCHIVAL_STATE_ENABLED,
namespaceDefaultURI: "some-uri",
}
}
// ClusterConfiguredForArchival returns true if cluster is configured to handle archival, false otherwise
func (a *archivalConfig) ClusterConfiguredForArchival() bool {
return a.GetClusterState() == ArchivalEnabled
}
func (a *archivalConfig) GetClusterState() ArchivalState {
// Only check dynamic config when archival is enabled in static config.
// If archival is disabled in static config, there will be no provider section in the static config
// and the archiver provider can not create any archiver. Therefore, in that case,
// even dynamic config says archival is enabled, we should ignore that.
// Only when archival is enabled in static config, should we check if there's any difference between static config and dynamic config.
if a.staticClusterState != ArchivalEnabled {
return a.staticClusterState
}
dynamicStateStr := a.dynamicClusterState()
dynamicState, err := getClusterArchivalState(dynamicStateStr)
if err != nil {
return ArchivalDisabled
}
return dynamicState
}
func (a *archivalConfig) ReadEnabled() bool {
if !a.ClusterConfiguredForArchival() {
return false
}
return a.enableRead()
}
func (a *archivalConfig) GetNamespaceDefaultState() enumspb.ArchivalState {
return a.namespaceDefaultState
}
func (a *archivalConfig) GetNamespaceDefaultURI() string {
return a.namespaceDefaultURI
}
func getClusterArchivalState(str string) (ArchivalState, error) {
str = strings.TrimSpace(strings.ToLower(str))
switch str {
case "", config.ArchivalDisabled:
return ArchivalDisabled, nil
case config.ArchivalPaused:
return ArchivalPaused, nil
case config.ArchivalEnabled:
return ArchivalEnabled, nil
}
return ArchivalDisabled, fmt.Errorf("invalid archival state of %v for cluster, valid states are: {\"\", \"disabled\", \"paused\", \"enabled\"}", str)
}
func getNamespaceArchivalState(str string) (enumspb.ArchivalState, error) {
str = strings.TrimSpace(strings.ToLower(str))
switch str {
case "", config.ArchivalDisabled:
return enumspb.ARCHIVAL_STATE_DISABLED, nil
case config.ArchivalEnabled:
return enumspb.ARCHIVAL_STATE_ENABLED, nil
}
return enumspb.ARCHIVAL_STATE_DISABLED, fmt.Errorf("invalid archival state of %v for namespace, valid states are: {\"\", \"disabled\", \"enabled\"}", str)
}