Files
tasks/service/frontend/namespace_handler.go
pdoernerandGitHub 6bb0519e8f Refactor protobuf duration validation (#6631)
## What changed?
<!-- Describe what has changed in this PR -->
Refactored and normalized protobuf duration validation.
Added additional validation for durations with mismatched signs.
Removed `timer.validation.ValidateAndCapTimer` and will instead rely on
task dropping logic in mutable state for long duration timers.

## Why?
<!-- Tell your future self why have you made these changes -->
Durations with mismatched signs can be converted to Golang durations but
they cannot be serialized to a single string which causes an error when
trying to view histories with these durations.

## How did you test it?
<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
Existing tests

## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
Low risk that this additional validation rejects durations which were
previously accepted and processed without issue. It is our assumption
that temporal created SDKs are not sending these invalid durations. The
error when viewing history has only been reported by one customer using
a custom Haskell SDK.
2024-10-18 10:12:11 -07:00

983 lines
33 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 namespace_handler_mock.go
package frontend
import (
"context"
"fmt"
"time"
"github.com/pborman/uuid"
enumspb "go.temporal.io/api/enums/v1"
namespacepb "go.temporal.io/api/namespace/v1"
replicationpb "go.temporal.io/api/replication/v1"
"go.temporal.io/api/serviceerror"
"go.temporal.io/api/workflowservice/v1"
enumsspb "go.temporal.io/server/api/enums/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
"go.temporal.io/server/common"
"go.temporal.io/server/common/archiver"
"go.temporal.io/server/common/archiver/provider"
"go.temporal.io/server/common/clock"
"go.temporal.io/server/common/cluster"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/log/tag"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/persistence"
"go.temporal.io/server/common/primitives"
"go.temporal.io/server/common/primitives/timestamp"
"go.temporal.io/server/common/util"
"google.golang.org/protobuf/types/known/durationpb"
"google.golang.org/protobuf/types/known/timestamppb"
)
type (
// namespaceHandler implements the namespace-related APIs specified by the [workflowservice] package,
// such as registering, updating, and querying namespaces.
namespaceHandler struct {
logger log.Logger
metadataMgr persistence.MetadataManager
clusterMetadata cluster.Metadata
namespaceReplicator namespace.Replicator
namespaceAttrValidator *namespace.AttrValidatorImpl
archivalMetadata archiver.ArchivalMetadata
archiverProvider provider.ArchiverProvider
timeSource clock.TimeSource
config *Config
}
)
const (
maxReplicationHistorySize = 10
)
var (
// err indicating that this cluster is not the master, so cannot do namespace registration or update
errNotMasterCluster = serviceerror.NewInvalidArgument("Cluster is not master cluster, cannot do namespace registration or namespace update.")
errCannotDoNamespaceFailoverAndUpdate = serviceerror.NewInvalidArgument("Cannot set active cluster to current cluster when other parameters are set.")
errInvalidRetentionPeriod = serviceerror.NewInvalidArgument("A valid retention period is not set on request.")
errInvalidNamespaceStateUpdate = serviceerror.NewInvalidArgument("Invalid namespace state update.")
errCustomSearchAttributeFieldAlreadyAllocated = serviceerror.NewInvalidArgument("Custom search attribute field name already allocated.")
)
// newNamespaceHandler create a new namespace handler
func newNamespaceHandler(
logger log.Logger,
metadataMgr persistence.MetadataManager,
clusterMetadata cluster.Metadata,
namespaceReplicator namespace.Replicator,
archivalMetadata archiver.ArchivalMetadata,
archiverProvider provider.ArchiverProvider,
timeSource clock.TimeSource,
config *Config,
) *namespaceHandler {
return &namespaceHandler{
logger: logger,
metadataMgr: metadataMgr,
clusterMetadata: clusterMetadata,
namespaceReplicator: namespaceReplicator,
namespaceAttrValidator: namespace.NewAttrValidator(clusterMetadata),
archivalMetadata: archivalMetadata,
archiverProvider: archiverProvider,
timeSource: timeSource,
config: config,
}
}
// RegisterNamespace register a new namespace
//
//nolint:revive // cognitive complexity grandfathered
func (d *namespaceHandler) RegisterNamespace(
ctx context.Context,
registerRequest *workflowservice.RegisterNamespaceRequest,
) (*workflowservice.RegisterNamespaceResponse, error) {
if !d.clusterMetadata.IsGlobalNamespaceEnabled() {
if registerRequest.GetIsGlobalNamespace() {
return nil, serviceerror.NewInvalidArgument("Cannot register global namespace when not enabled")
}
registerRequest.IsGlobalNamespace = false
} else {
// cluster global namespace enabled
if !d.clusterMetadata.IsMasterCluster() && registerRequest.GetIsGlobalNamespace() {
return nil, errNotMasterCluster
}
}
if err := validateRetentionDuration(
registerRequest.WorkflowExecutionRetentionPeriod,
registerRequest.IsGlobalNamespace,
); err != nil {
return nil, err
}
// first check if the name is already registered as the local namespace
_, err := d.metadataMgr.GetNamespace(ctx, &persistence.GetNamespaceRequest{Name: registerRequest.GetNamespace()})
switch err.(type) {
case nil:
// namespace already exists, cannot proceed
return nil, serviceerror.NewNamespaceAlreadyExists("Namespace already exists.")
case *serviceerror.NamespaceNotFound:
// namespace does not exists, proceeds
default:
// other err
return nil, err
}
var activeClusterName string
// input validation on cluster names
if registerRequest.GetActiveClusterName() != "" {
activeClusterName = registerRequest.GetActiveClusterName()
} else {
activeClusterName = d.clusterMetadata.GetCurrentClusterName()
}
var clusters []string
for _, clusterConfig := range registerRequest.Clusters {
clusterName := clusterConfig.GetClusterName()
clusters = append(clusters, clusterName)
}
clusters = persistence.GetOrUseDefaultClusters(activeClusterName, clusters)
currentHistoryArchivalState := namespace.NeverEnabledState()
nextHistoryArchivalState := currentHistoryArchivalState
clusterHistoryArchivalConfig := d.archivalMetadata.GetHistoryConfig()
if clusterHistoryArchivalConfig.ClusterConfiguredForArchival() {
archivalEvent, err := d.toArchivalRegisterEvent(
registerRequest.HistoryArchivalState,
registerRequest.GetHistoryArchivalUri(),
clusterHistoryArchivalConfig.GetNamespaceDefaultState(),
clusterHistoryArchivalConfig.GetNamespaceDefaultURI(),
)
if err != nil {
return nil, err
}
nextHistoryArchivalState, _, err = currentHistoryArchivalState.GetNextState(archivalEvent, d.validateHistoryArchivalURI)
if err != nil {
return nil, err
}
}
currentVisibilityArchivalState := namespace.NeverEnabledState()
nextVisibilityArchivalState := currentVisibilityArchivalState
clusterVisibilityArchivalConfig := d.archivalMetadata.GetVisibilityConfig()
if clusterVisibilityArchivalConfig.ClusterConfiguredForArchival() {
archivalEvent, err := d.toArchivalRegisterEvent(
registerRequest.VisibilityArchivalState,
registerRequest.GetVisibilityArchivalUri(),
clusterVisibilityArchivalConfig.GetNamespaceDefaultState(),
clusterVisibilityArchivalConfig.GetNamespaceDefaultURI(),
)
if err != nil {
return nil, err
}
nextVisibilityArchivalState, _, err = currentVisibilityArchivalState.GetNextState(archivalEvent, d.validateVisibilityArchivalURI)
if err != nil {
return nil, err
}
}
info := &persistencespb.NamespaceInfo{
Id: uuid.New(),
Name: registerRequest.GetNamespace(),
State: enumspb.NAMESPACE_STATE_REGISTERED,
Owner: registerRequest.GetOwnerEmail(),
Description: registerRequest.GetDescription(),
Data: registerRequest.Data,
}
config := &persistencespb.NamespaceConfig{
Retention: registerRequest.GetWorkflowExecutionRetentionPeriod(),
HistoryArchivalState: nextHistoryArchivalState.State,
HistoryArchivalUri: nextHistoryArchivalState.URI,
VisibilityArchivalState: nextVisibilityArchivalState.State,
VisibilityArchivalUri: nextVisibilityArchivalState.URI,
BadBinaries: &namespacepb.BadBinaries{Binaries: map[string]*namespacepb.BadBinaryInfo{}},
CustomSearchAttributeAliases: nil,
}
replicationConfig := &persistencespb.NamespaceReplicationConfig{
ActiveClusterName: activeClusterName,
Clusters: clusters,
State: enumspb.REPLICATION_STATE_NORMAL,
}
isGlobalNamespace := registerRequest.GetIsGlobalNamespace()
if err := d.namespaceAttrValidator.ValidateNamespaceConfig(config); err != nil {
return nil, err
}
if isGlobalNamespace {
if err := d.namespaceAttrValidator.ValidateNamespaceReplicationConfigForGlobalNamespace(
replicationConfig,
); err != nil {
return nil, err
}
} else {
if err := d.namespaceAttrValidator.ValidateNamespaceReplicationConfigForLocalNamespace(
replicationConfig,
); err != nil {
return nil, err
}
}
failoverVersion := common.EmptyVersion
if registerRequest.GetIsGlobalNamespace() {
failoverVersion = d.clusterMetadata.GetNextFailoverVersion(activeClusterName, 0)
}
namespaceRequest := &persistence.CreateNamespaceRequest{
Namespace: &persistencespb.NamespaceDetail{
Info: info,
Config: config,
ReplicationConfig: replicationConfig,
ConfigVersion: 0,
FailoverVersion: failoverVersion,
},
IsGlobalNamespace: isGlobalNamespace,
}
namespaceResponse, err := d.metadataMgr.CreateNamespace(ctx, namespaceRequest)
if err != nil {
return nil, err
}
err = d.namespaceReplicator.HandleTransmissionTask(
ctx,
enumsspb.NAMESPACE_OPERATION_CREATE,
namespaceRequest.Namespace.Info,
namespaceRequest.Namespace.Config,
namespaceRequest.Namespace.ReplicationConfig,
false,
namespaceRequest.Namespace.ConfigVersion,
namespaceRequest.Namespace.FailoverVersion,
namespaceRequest.IsGlobalNamespace,
nil,
)
if err != nil {
return nil, err
}
d.logger.Info("Register namespace succeeded",
tag.WorkflowNamespace(registerRequest.GetNamespace()),
tag.WorkflowNamespaceID(namespaceResponse.ID),
)
return &workflowservice.RegisterNamespaceResponse{}, nil
}
// ListNamespaces list all namespaces
func (d *namespaceHandler) ListNamespaces(
ctx context.Context,
listRequest *workflowservice.ListNamespacesRequest,
) (*workflowservice.ListNamespacesResponse, error) {
pageSize := 100
if listRequest.GetPageSize() != 0 {
pageSize = int(listRequest.GetPageSize())
}
resp, err := d.metadataMgr.ListNamespaces(ctx, &persistence.ListNamespacesRequest{
PageSize: pageSize,
NextPageToken: listRequest.NextPageToken,
IncludeDeleted: listRequest.GetNamespaceFilter().GetIncludeDeleted(),
})
if err != nil {
return nil, err
}
var namespaces []*workflowservice.DescribeNamespaceResponse
for _, namespace := range resp.Namespaces {
desc := &workflowservice.DescribeNamespaceResponse{
IsGlobalNamespace: namespace.IsGlobalNamespace,
FailoverVersion: namespace.Namespace.FailoverVersion,
}
desc.NamespaceInfo, desc.Config, desc.ReplicationConfig, desc.FailoverHistory =
d.createResponse(
namespace.Namespace.Info,
namespace.Namespace.Config,
namespace.Namespace.ReplicationConfig)
namespaces = append(namespaces, desc)
}
response := &workflowservice.ListNamespacesResponse{
Namespaces: namespaces,
NextPageToken: resp.NextPageToken,
}
return response, nil
}
// DescribeNamespace describe the namespace
func (d *namespaceHandler) DescribeNamespace(
ctx context.Context,
describeRequest *workflowservice.DescribeNamespaceRequest,
) (*workflowservice.DescribeNamespaceResponse, error) {
// TODO, we should migrate the non global namespace to new table, see #773
req := &persistence.GetNamespaceRequest{
Name: describeRequest.GetNamespace(),
ID: describeRequest.GetId(),
}
resp, err := d.metadataMgr.GetNamespace(ctx, req)
if err != nil {
return nil, err
}
response := &workflowservice.DescribeNamespaceResponse{
IsGlobalNamespace: resp.IsGlobalNamespace,
FailoverVersion: resp.Namespace.FailoverVersion,
}
response.NamespaceInfo, response.Config, response.ReplicationConfig, response.FailoverHistory =
d.createResponse(resp.Namespace.Info, resp.Namespace.Config, resp.Namespace.ReplicationConfig)
return response, nil
}
// UpdateNamespace update the namespace
//
//nolint:revive // cognitive complexity grandfathered
func (d *namespaceHandler) UpdateNamespace(
ctx context.Context,
updateRequest *workflowservice.UpdateNamespaceRequest,
) (*workflowservice.UpdateNamespaceResponse, error) {
// must get the metadata (notificationVersion) first
// this version can be regarded as the lock on the v2 namespace table
// and since we do not know which table will return the namespace afterwards
// this call has to be made
metadata, err := d.metadataMgr.GetMetadata(ctx)
if err != nil {
return nil, err
}
notificationVersion := metadata.NotificationVersion
getResponse, err := d.metadataMgr.GetNamespace(ctx, &persistence.GetNamespaceRequest{Name: updateRequest.GetNamespace()})
if err != nil {
return nil, err
}
info := getResponse.Namespace.Info
config := getResponse.Namespace.Config
replicationConfig := getResponse.Namespace.ReplicationConfig
failoverHistory := getResponse.Namespace.ReplicationConfig.FailoverHistory
configVersion := getResponse.Namespace.ConfigVersion
failoverVersion := getResponse.Namespace.FailoverVersion
failoverNotificationVersion := getResponse.Namespace.FailoverNotificationVersion
isGlobalNamespace := getResponse.IsGlobalNamespace || updateRequest.PromoteNamespace
needsNamespacePromotion := !getResponse.IsGlobalNamespace && updateRequest.PromoteNamespace
currentHistoryArchivalState := &namespace.ArchivalConfigState{
State: config.HistoryArchivalState,
URI: config.HistoryArchivalUri,
}
nextHistoryArchivalState := currentHistoryArchivalState
historyArchivalConfigChanged := false
clusterHistoryArchivalConfig := d.archivalMetadata.GetHistoryConfig()
if updateRequest.Config != nil && clusterHistoryArchivalConfig.ClusterConfiguredForArchival() {
cfg := updateRequest.GetConfig()
archivalEvent, err := d.toArchivalUpdateEvent(cfg.HistoryArchivalState, cfg.GetHistoryArchivalUri(), clusterHistoryArchivalConfig.GetNamespaceDefaultURI())
if err != nil {
return nil, err
}
nextHistoryArchivalState, historyArchivalConfigChanged, err = currentHistoryArchivalState.GetNextState(archivalEvent, d.validateHistoryArchivalURI)
if err != nil {
return nil, err
}
}
currentVisibilityArchivalState := &namespace.ArchivalConfigState{
State: config.VisibilityArchivalState,
URI: config.VisibilityArchivalUri,
}
nextVisibilityArchivalState := currentVisibilityArchivalState
visibilityArchivalConfigChanged := false
clusterVisibilityArchivalConfig := d.archivalMetadata.GetVisibilityConfig()
if updateRequest.Config != nil && clusterVisibilityArchivalConfig.ClusterConfiguredForArchival() {
cfg := updateRequest.GetConfig()
archivalEvent, err := d.toArchivalUpdateEvent(cfg.VisibilityArchivalState, cfg.GetVisibilityArchivalUri(), clusterVisibilityArchivalConfig.GetNamespaceDefaultURI())
if err != nil {
return nil, err
}
nextVisibilityArchivalState, visibilityArchivalConfigChanged, err = currentVisibilityArchivalState.GetNextState(archivalEvent, d.validateVisibilityArchivalURI)
if err != nil {
return nil, err
}
}
// whether active cluster is changed
activeClusterChanged := false
// whether anything other than active cluster is changed
configurationChanged := false
// whether replication cluster list is changed
clusterListChanged := false
if updateRequest.UpdateInfo != nil {
updatedInfo := updateRequest.UpdateInfo
if updatedInfo.GetDescription() != "" {
configurationChanged = true
info.Description = updatedInfo.GetDescription()
}
if updatedInfo.GetOwnerEmail() != "" {
configurationChanged = true
info.Owner = updatedInfo.GetOwnerEmail()
}
if updatedInfo.Data != nil {
configurationChanged = true
// only do merging
info.Data = d.mergeNamespaceData(info.Data, updatedInfo.Data)
}
if updatedInfo.State != enumspb.NAMESPACE_STATE_UNSPECIFIED && info.State != updatedInfo.State {
configurationChanged = true
if err := validateStateUpdate(getResponse, updateRequest); err != nil {
return nil, err
}
info.State = updatedInfo.State
}
}
if updateRequest.Config != nil {
updatedConfig := updateRequest.Config
if updatedConfig.GetWorkflowExecutionRetentionTtl() != nil {
configurationChanged = true
config.Retention = updatedConfig.GetWorkflowExecutionRetentionTtl()
if err := validateRetentionDuration(
config.Retention,
isGlobalNamespace,
); err != nil {
return nil, err
}
}
if historyArchivalConfigChanged {
configurationChanged = true
config.HistoryArchivalState = nextHistoryArchivalState.State
config.HistoryArchivalUri = nextHistoryArchivalState.URI
}
if visibilityArchivalConfigChanged {
configurationChanged = true
config.VisibilityArchivalState = nextVisibilityArchivalState.State
config.VisibilityArchivalUri = nextVisibilityArchivalState.URI
}
if updatedConfig.BadBinaries != nil {
maxLength := d.config.MaxBadBinaries(updateRequest.GetNamespace())
// only do merging
bb := d.mergeBadBinaries(config.BadBinaries.Binaries, updatedConfig.BadBinaries.Binaries, time.Now().UTC())
config.BadBinaries = &bb
if len(config.BadBinaries.Binaries) > maxLength {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("Total resetBinaries cannot exceed the max limit: %v", maxLength))
}
}
if len(updatedConfig.CustomSearchAttributeAliases) > 0 {
configurationChanged = true
csaAliases, err := d.upsertCustomSearchAttributesAliases(
config.CustomSearchAttributeAliases,
updatedConfig.CustomSearchAttributeAliases,
)
if err != nil {
return nil, err
}
config.CustomSearchAttributeAliases = csaAliases
}
}
if updateRequest.GetDeleteBadBinary() != "" {
binChecksum := updateRequest.GetDeleteBadBinary()
_, ok := config.BadBinaries.Binaries[binChecksum]
if !ok {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("Bad binary checksum %v doesn't exists.", binChecksum))
}
configurationChanged = true
delete(config.BadBinaries.Binaries, binChecksum)
}
if updateRequest.ReplicationConfig != nil {
updateReplicationConfig := updateRequest.ReplicationConfig
if len(updateReplicationConfig.Clusters) != 0 {
configurationChanged = true
clusterListChanged = true
var clustersNew []string
for _, clusterConfig := range updateReplicationConfig.Clusters {
clustersNew = append(clustersNew, clusterConfig.GetClusterName())
}
replicationConfig.Clusters = clustersNew
}
if updateReplicationConfig.State != enumspb.REPLICATION_STATE_UNSPECIFIED &&
updateReplicationConfig.State != replicationConfig.State {
if err := validateReplicationStateUpdate(getResponse, updateRequest); err != nil {
return nil, err
}
configurationChanged = true
replicationConfig.State = updateReplicationConfig.State
}
if updateReplicationConfig.GetActiveClusterName() != "" {
activeClusterChanged = true
replicationConfig.ActiveClusterName = updateReplicationConfig.GetActiveClusterName()
}
}
if err := d.namespaceAttrValidator.ValidateNamespaceConfig(config); err != nil {
return nil, err
}
if isGlobalNamespace {
if err := d.namespaceAttrValidator.ValidateNamespaceReplicationConfigForGlobalNamespace(
replicationConfig,
); err != nil {
return nil, err
}
if !d.clusterMetadata.IsGlobalNamespaceEnabled() {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("global namespace is not enabled on this "+
"cluster, cannot update global namespace or promote local namespace: %v", updateRequest.Namespace))
}
} else {
if err := d.namespaceAttrValidator.ValidateNamespaceReplicationConfigForLocalNamespace(
replicationConfig,
); err != nil {
return nil, err
}
}
if configurationChanged && activeClusterChanged && isGlobalNamespace {
return nil, errCannotDoNamespaceFailoverAndUpdate
} else if configurationChanged || activeClusterChanged || needsNamespacePromotion {
if (needsNamespacePromotion || activeClusterChanged) && isGlobalNamespace {
failoverVersion = d.clusterMetadata.GetNextFailoverVersion(
replicationConfig.ActiveClusterName,
failoverVersion,
)
failoverNotificationVersion = notificationVersion
}
// set the versions
if configurationChanged {
configVersion++
}
if (configurationChanged || activeClusterChanged) && isGlobalNamespace {
// N.B., it should be sufficient to check only for activeClusterChanged. In order to be defensive, we also
// check for configurationChanged. If nothing needs to be updated this will be a no-op.
failoverHistory = d.maybeUpdateFailoverHistory(
failoverHistory,
updateRequest.ReplicationConfig,
getResponse.Namespace,
failoverVersion,
)
}
replicationConfig.FailoverHistory = failoverHistory
updateReq := &persistence.UpdateNamespaceRequest{
Namespace: &persistencespb.NamespaceDetail{
Info: info,
Config: config,
ReplicationConfig: replicationConfig,
ConfigVersion: configVersion,
FailoverVersion: failoverVersion,
FailoverNotificationVersion: failoverNotificationVersion,
},
IsGlobalNamespace: isGlobalNamespace,
NotificationVersion: notificationVersion,
}
err = d.metadataMgr.UpdateNamespace(ctx, updateReq)
if err != nil {
return nil, err
}
}
err = d.namespaceReplicator.HandleTransmissionTask(
ctx,
enumsspb.NAMESPACE_OPERATION_UPDATE,
info,
config,
replicationConfig,
clusterListChanged,
configVersion,
failoverVersion,
isGlobalNamespace,
failoverHistory,
)
if err != nil {
return nil, err
}
response := &workflowservice.UpdateNamespaceResponse{
IsGlobalNamespace: isGlobalNamespace,
FailoverVersion: failoverVersion,
}
response.NamespaceInfo, response.Config, response.ReplicationConfig, _ = d.createResponse(info, config, replicationConfig)
d.logger.Info("Update namespace succeeded",
tag.WorkflowNamespace(info.Name),
tag.WorkflowNamespaceID(info.Id),
)
return response, nil
}
// DeprecateNamespace deprecates a namespace
// Deprecated.
func (d *namespaceHandler) DeprecateNamespace(
ctx context.Context,
deprecateRequest *workflowservice.DeprecateNamespaceRequest,
) (*workflowservice.DeprecateNamespaceResponse, error) {
clusterMetadata := d.clusterMetadata
// TODO remove the IsGlobalNamespaceEnabled check once cross DC is public
if clusterMetadata.IsGlobalNamespaceEnabled() && !clusterMetadata.IsMasterCluster() {
return nil, errNotMasterCluster
}
// must get the metadata (notificationVersion) first
// this version can be regarded as the lock on the v2 namespace table
// and since we do not know which table will return the namespace afterwards
// this call has to be made
metadata, err := d.metadataMgr.GetMetadata(ctx)
if err != nil {
return nil, err
}
notificationVersion := metadata.NotificationVersion
getResponse, err := d.metadataMgr.GetNamespace(ctx, &persistence.GetNamespaceRequest{Name: deprecateRequest.GetNamespace()})
if err != nil {
return nil, err
}
getResponse.Namespace.ConfigVersion = getResponse.Namespace.ConfigVersion + 1
getResponse.Namespace.Info.State = enumspb.NAMESPACE_STATE_DEPRECATED
updateReq := &persistence.UpdateNamespaceRequest{
Namespace: &persistencespb.NamespaceDetail{
Info: getResponse.Namespace.Info,
Config: getResponse.Namespace.Config,
ReplicationConfig: getResponse.Namespace.ReplicationConfig,
ConfigVersion: getResponse.Namespace.ConfigVersion,
FailoverVersion: getResponse.Namespace.FailoverVersion,
FailoverNotificationVersion: getResponse.Namespace.FailoverNotificationVersion,
},
NotificationVersion: notificationVersion,
IsGlobalNamespace: getResponse.IsGlobalNamespace,
}
err = d.metadataMgr.UpdateNamespace(ctx, updateReq)
if err != nil {
return nil, err
}
return nil, nil
}
func (d *namespaceHandler) createResponse(
info *persistencespb.NamespaceInfo,
config *persistencespb.NamespaceConfig,
replicationConfig *persistencespb.NamespaceReplicationConfig,
) (*namespacepb.NamespaceInfo, *namespacepb.NamespaceConfig, *replicationpb.NamespaceReplicationConfig, []*replicationpb.FailoverStatus) {
infoResult := &namespacepb.NamespaceInfo{
Name: info.Name,
State: info.State,
Description: info.Description,
OwnerEmail: info.Owner,
Data: info.Data,
Id: info.Id,
Capabilities: &namespacepb.NamespaceInfo_Capabilities{
EagerWorkflowStart: d.config.EnableEagerWorkflowStart(info.Name),
SyncUpdate: d.config.EnableUpdateWorkflowExecution(info.Name),
AsyncUpdate: d.config.EnableUpdateWorkflowExecutionAsyncAccepted(info.Name),
},
SupportsSchedules: d.config.EnableSchedules(info.Name),
}
configResult := &namespacepb.NamespaceConfig{
WorkflowExecutionRetentionTtl: config.Retention,
HistoryArchivalState: config.HistoryArchivalState,
HistoryArchivalUri: config.HistoryArchivalUri,
VisibilityArchivalState: config.VisibilityArchivalState,
VisibilityArchivalUri: config.VisibilityArchivalUri,
BadBinaries: config.BadBinaries,
CustomSearchAttributeAliases: config.CustomSearchAttributeAliases,
}
var clusters []*replicationpb.ClusterReplicationConfig
for _, cluster := range replicationConfig.Clusters {
clusters = append(clusters, &replicationpb.ClusterReplicationConfig{
ClusterName: cluster,
})
}
replicationConfigResult := &replicationpb.NamespaceReplicationConfig{
ActiveClusterName: replicationConfig.ActiveClusterName,
Clusters: clusters,
}
var failoverHistory []*replicationpb.FailoverStatus
for _, entry := range replicationConfig.GetFailoverHistory() {
failoverHistory = append(failoverHistory, &replicationpb.FailoverStatus{
FailoverTime: entry.GetFailoverTime(),
FailoverVersion: entry.GetFailoverVersion(),
})
}
return infoResult, configResult, replicationConfigResult, failoverHistory
}
func (d *namespaceHandler) mergeBadBinaries(
old map[string]*namespacepb.BadBinaryInfo,
new map[string]*namespacepb.BadBinaryInfo,
createTime time.Time,
) namespacepb.BadBinaries {
if old == nil {
old = map[string]*namespacepb.BadBinaryInfo{}
}
for k, v := range new {
v.CreateTime = timestamppb.New(createTime)
old[k] = v
}
return namespacepb.BadBinaries{
Binaries: old,
}
}
func (d *namespaceHandler) mergeNamespaceData(
old map[string]string,
new map[string]string,
) map[string]string {
if old == nil {
old = map[string]string{}
}
for k, v := range new {
old[k] = v
}
return old
}
func (d *namespaceHandler) upsertCustomSearchAttributesAliases(
current map[string]string,
upsert map[string]string,
) (map[string]string, error) {
result := util.CloneMapNonNil(current)
for key, value := range upsert {
if value == "" {
delete(result, key)
} else if _, ok := current[key]; !ok {
result[key] = value
} else {
return nil, errCustomSearchAttributeFieldAlreadyAllocated
}
}
return result, nil
}
func (d *namespaceHandler) toArchivalRegisterEvent(
state enumspb.ArchivalState,
URI string,
defaultState enumspb.ArchivalState,
defaultURI string,
) (*namespace.ArchivalConfigEvent, error) {
event := &namespace.ArchivalConfigEvent{
State: state,
URI: URI,
DefaultURI: defaultURI,
}
if event.State == enumspb.ARCHIVAL_STATE_UNSPECIFIED {
event.State = defaultState
}
if err := event.Validate(); err != nil {
return nil, err
}
return event, nil
}
func (d *namespaceHandler) toArchivalUpdateEvent(
state enumspb.ArchivalState,
URI string,
defaultURI string,
) (*namespace.ArchivalConfigEvent, error) {
event := &namespace.ArchivalConfigEvent{
State: state,
URI: URI,
DefaultURI: defaultURI,
}
if err := event.Validate(); err != nil {
return nil, err
}
return event, nil
}
func (d *namespaceHandler) validateHistoryArchivalURI(URIString string) error {
URI, err := archiver.NewURI(URIString)
if err != nil {
return err
}
archiver, err := d.archiverProvider.GetHistoryArchiver(URI.Scheme(), string(primitives.FrontendService))
if err != nil {
return err
}
return archiver.ValidateURI(URI)
}
func (d *namespaceHandler) validateVisibilityArchivalURI(URIString string) error {
URI, err := archiver.NewURI(URIString)
if err != nil {
return err
}
archiver, err := d.archiverProvider.GetVisibilityArchiver(URI.Scheme(), string(primitives.FrontendService))
if err != nil {
return err
}
return archiver.ValidateURI(URI)
}
// maybeUpdateFailoverHistory adds an entry if the Namespace is becoming active in a new cluster.
func (d *namespaceHandler) maybeUpdateFailoverHistory(
failoverHistory []*persistencespb.FailoverStatus,
updateReplicationConfig *replicationpb.NamespaceReplicationConfig,
namespaceDetail *persistencespb.NamespaceDetail,
newFailoverVersion int64,
) []*persistencespb.FailoverStatus {
d.logger.Debug(
"maybeUpdateFailoverHistory",
tag.NewAnyTag("failoverHistory", failoverHistory),
tag.NewAnyTag("updateReplConfig", updateReplicationConfig),
tag.NewAnyTag("namespaceDetail", namespaceDetail),
)
if updateReplicationConfig == nil {
d.logger.Debug("updateReplicationConfig was nil")
return failoverHistory
}
lastFailoverVersion := int64(-1)
if l := len(namespaceDetail.ReplicationConfig.FailoverHistory); l > 0 {
lastFailoverVersion = namespaceDetail.ReplicationConfig.FailoverHistory[l-1].FailoverVersion
}
if lastFailoverVersion != newFailoverVersion {
now := d.timeSource.Now()
failoverHistory = append(
failoverHistory, &persistencespb.FailoverStatus{
FailoverTime: timestamppb.New(now),
FailoverVersion: newFailoverVersion,
},
)
}
if l := len(failoverHistory); l > maxReplicationHistorySize {
failoverHistory = failoverHistory[l-maxReplicationHistorySize : l]
}
return failoverHistory
}
// validateRetentionDuration ensures that retention duration can't be set below a sane minimum.
func validateRetentionDuration(retention *durationpb.Duration, isGlobalNamespace bool) error {
if err := timestamp.ValidateProtoDuration(retention); err != nil {
return errInvalidRetentionPeriod
}
min := namespace.MinRetentionLocal
if isGlobalNamespace {
min = namespace.MinRetentionGlobal
}
if timestamp.DurationValue(retention) < min {
return errInvalidRetentionPeriod
}
return nil
}
func validateReplicationStateUpdate(existingNamespace *persistence.GetNamespaceResponse, nsUpdateRequest *workflowservice.UpdateNamespaceRequest) error {
if nsUpdateRequest.ReplicationConfig == nil ||
nsUpdateRequest.ReplicationConfig.State == enumspb.REPLICATION_STATE_UNSPECIFIED ||
nsUpdateRequest.ReplicationConfig.State == existingNamespace.Namespace.ReplicationConfig.State {
return nil // no change
}
if existingNamespace.Namespace.Info.State != enumspb.NAMESPACE_STATE_REGISTERED {
return serviceerror.NewInvalidArgument(
fmt.Sprintf(
"update ReplicationState is only supported when namespace is in %s state, current state: %s",
enumspb.NAMESPACE_STATE_REGISTERED.String(),
existingNamespace.Namespace.Info.State.String(),
),
)
}
if nsUpdateRequest.ReplicationConfig.State == enumspb.REPLICATION_STATE_HANDOVER {
if !existingNamespace.IsGlobalNamespace {
return serviceerror.NewInvalidArgument(
fmt.Sprintf(
"%s can only be set for global namespace",
enumspb.REPLICATION_STATE_HANDOVER,
),
)
}
// verify namespace has more than 1 replication clusters
replicationClusterCount := len(existingNamespace.Namespace.ReplicationConfig.Clusters)
if len(nsUpdateRequest.ReplicationConfig.Clusters) > 0 {
replicationClusterCount = len(nsUpdateRequest.ReplicationConfig.Clusters)
}
if replicationClusterCount < 2 {
return serviceerror.NewInvalidArgument(fmt.Sprintf("%s require more than one replication clusters", enumspb.REPLICATION_STATE_HANDOVER))
}
}
return nil
}
func validateStateUpdate(existingNamespace *persistence.GetNamespaceResponse, nsUpdateRequest *workflowservice.UpdateNamespaceRequest) error {
if nsUpdateRequest.UpdateInfo == nil {
return nil // no change
}
oldState := existingNamespace.Namespace.Info.State
newState := nsUpdateRequest.UpdateInfo.State
if newState == enumspb.NAMESPACE_STATE_UNSPECIFIED || oldState == newState {
return nil // no change
}
if existingNamespace.Namespace.ReplicationConfig != nil &&
existingNamespace.Namespace.ReplicationConfig.State == enumspb.REPLICATION_STATE_HANDOVER {
return serviceerror.NewInvalidArgument("cannot update namespace state while its replication state in REPLICATION_STATE_HANDOVER")
}
switch oldState {
case enumspb.NAMESPACE_STATE_REGISTERED:
switch newState {
case enumspb.NAMESPACE_STATE_DELETED, enumspb.NAMESPACE_STATE_DEPRECATED:
return nil
default:
return errInvalidNamespaceStateUpdate
}
case enumspb.NAMESPACE_STATE_DEPRECATED:
switch newState {
case enumspb.NAMESPACE_STATE_DELETED:
return nil
default:
return errInvalidNamespaceStateUpdate
}
case enumspb.NAMESPACE_STATE_DELETED:
return errInvalidNamespaceStateUpdate
default:
return errInvalidNamespaceStateUpdate
}
}