Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Delete Namespace: read cache refresh interval from DC #7215

Merged
merged 1 commit into from
Feb 3, 2025
Merged

Delete Namespace: read cache refresh interval from DC #7215

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion service/worker/deletenamespace/fx.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ type (
allowDeleteNamespaceIfNexusEndpointTarget dynamicconfig.BoolPropertyFn
nexusEndpointListDefaultPageSize dynamicconfig.IntPropertyFn
deleteActivityRPS dynamicconfig.TypedSubscribable[int]
namespaceCacheRefreshInterval dynamicconfig.DurationPropertyFn
}
componentParams struct {
fx.In
Expand Down Expand Up @@ -92,6 +93,7 @@ func newComponent(
allowDeleteNamespaceIfNexusEndpointTarget: dynamicconfig.AllowDeleteNamespaceIfNexusEndpointTarget.Get(params.DynamicCollection),
nexusEndpointListDefaultPageSize: dynamicconfig.NexusEndpointListDefaultPageSize.Get(params.DynamicCollection),
deleteActivityRPS: dynamicconfig.DeleteNamespaceDeleteActivityRPS.Subscribe(params.DynamicCollection),
namespaceCacheRefreshInterval: dynamicconfig.NamespaceCacheRefreshInterval.Get(params.DynamicCollection),
}
}

Expand Down Expand Up @@ -145,7 +147,7 @@ func (wc *deleteNamespaceComponent) reclaimResourcesActivities() *reclaimresourc
}

func (wc *deleteNamespaceComponent) reclaimResourcesLocalActivities() *reclaimresources.LocalActivities {
return reclaimresources.NewLocalActivities(wc.visibilityManager, wc.metadataManager, wc.logger)
return reclaimresources.NewLocalActivities(wc.visibilityManager, wc.metadataManager, wc.namespaceCacheRefreshInterval, wc.logger)
}

func (wc *deleteNamespaceComponent) deleteExecutionsActivities() *deleteexecutions.Activities {
Expand Down
17 changes: 13 additions & 4 deletions service/worker/deletenamespace/reclaimresources/activities.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,13 @@ package reclaimresources

import (
"context"
"time"

"go.temporal.io/sdk/activity"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/headers"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/log/tag"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/persistence"
"go.temporal.io/server/common/persistence/visibility/manager"
Expand All @@ -42,15 +43,16 @@ import (
type (
Activities struct {
visibilityManager manager.VisibilityManager
metricsHandler metrics.Handler
logger log.Logger
}

LocalActivities struct {
visibilityManager manager.VisibilityManager
metadataManager persistence.MetadataManager
metricsHandler metrics.Handler
logger log.Logger

namespaceCacheRefreshInterval dynamicconfig.DurationPropertyFn

logger log.Logger
}
)

Expand All @@ -67,12 +69,15 @@ func NewActivities(
func NewLocalActivities(
visibilityManager manager.VisibilityManager,
metadataManager persistence.MetadataManager,
namespaceCacheRefreshInterval dynamicconfig.DurationPropertyFn,
logger log.Logger,
) *LocalActivities {
return &LocalActivities{
visibilityManager: visibilityManager,
metadataManager: metadataManager,
logger: logger,

namespaceCacheRefreshInterval: namespaceCacheRefreshInterval,
}
}

Expand Down Expand Up @@ -176,3 +181,7 @@ func (a *LocalActivities) DeleteNamespaceActivity(ctx context.Context, nsID name
logger.Info("Namespace is deleted.")
return nil
}

func (a *LocalActivities) GetNamespaceCacheRefreshInterval(_ context.Context) (time.Duration, error) {
return a.namespaceCacheRefreshInterval(), nil
}
4 changes: 0 additions & 4 deletions service/worker/deletenamespace/reclaimresources/activities_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@ import (
"go.temporal.io/sdk/temporal"
"go.temporal.io/sdk/testsuite"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/persistence/visibility/manager"
"go.temporal.io/server/common/searchattribute"
Expand All @@ -53,7 +52,6 @@ func Test_EnsureNoExecutionsAdvVisibilityActivity_NoExecutions(t *testing.T) {

a := &Activities{
visibilityManager: visibilityManager,
metricsHandler: metrics.NoopMetricsHandler,
logger: log.NewTestLogger(),
}

Expand All @@ -78,7 +76,6 @@ func Test_EnsureNoExecutionsAdvVisibilityActivity_ExecutionsExist(t *testing.T)

a := &Activities{
visibilityManager: visibilityManager,
metricsHandler: metrics.NoopMetricsHandler,
logger: log.NewTestLogger(),
}
env.RegisterActivity(a.EnsureNoExecutionsAdvVisibilityActivity)
Expand Down Expand Up @@ -107,7 +104,6 @@ func Test_EnsureNoExecutionsAdvVisibilityActivity_NotDeletedExecutionsExist(t *t

a := &Activities{
visibilityManager: visibilityManager,
metricsHandler: metrics.NoopMetricsHandler,
logger: log.NewTestLogger(),
}
env.RegisterActivity(a.EnsureNoExecutionsAdvVisibilityActivity)
Expand Down
13 changes: 9 additions & 4 deletions service/worker/deletenamespace/reclaimresources/workflow.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,6 @@ import (

const (
WorkflowName = "temporal-sys-reclaim-namespace-resources-workflow"

namespaceCacheRefreshDelay = 11 * time.Second
)

type (
Expand Down Expand Up @@ -180,8 +178,15 @@ func ReclaimResourcesWorkflow(ctx workflow.Context, params ReclaimResourcesParam
var la *LocalActivities

// Step 0. This workflow is started right after the namespace is marked as DELETED and renamed.
// Wait for namespace cache refresh to make sure no new executions are created.
err = workflow.Sleep(ctx, namespaceCacheRefreshDelay)
// Wait for namespace cache refresh to make sure no new executions are created. 2 secodnds is a random buffer.
ctx0 := workflow.WithLocalActivityOptions(ctx, localActivityOptions)
var namespaceCacheRefreshDelay time.Duration
err = workflow.ExecuteLocalActivity(ctx0, la.GetNamespaceCacheRefreshInterval).Get(ctx, &namespaceCacheRefreshDelay)
if err != nil {
return result, err
}

err = workflow.Sleep(ctx, namespaceCacheRefreshDelay+2*time.Second)
if err != nil {
return result, err
}
Expand Down
25 changes: 15 additions & 10 deletions service/worker/deletenamespace/reclaimresources/workflow_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,8 @@ import (
"github.com/stretchr/testify/require"
"go.temporal.io/sdk/temporal"
"go.temporal.io/sdk/testsuite"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/persistence"
"go.temporal.io/server/common/persistence/visibility/manager"
Expand Down Expand Up @@ -70,6 +70,7 @@ func Test_ReclaimResourcesWorkflow_Success(t *testing.T) {
ErrorCount: 0,
}, nil).Once()

env.OnActivity(la.GetNamespaceCacheRefreshInterval, mock.Anything).Return(10*time.Second, nil).Once()
env.OnActivity(la.CountExecutionsAdvVisibilityActivity, mock.Anything, namespace.ID("namespace-id"), namespace.Name("namespace")).Return(int64(10), nil).Once()
env.OnActivity(a.EnsureNoExecutionsAdvVisibilityActivity, mock.Anything, namespace.ID("namespace-id"), namespace.Name("namespace"), 0).Return(nil).Once()

Expand Down Expand Up @@ -120,6 +121,7 @@ func Test_ReclaimResourcesWorkflow_EnsureNoExecutionsActivity_Error(t *testing.T
ErrorCount: 0,
}, nil).Once()

env.OnActivity(la.GetNamespaceCacheRefreshInterval, mock.Anything).Return(10*time.Second, nil).Once()
env.OnActivity(la.CountExecutionsAdvVisibilityActivity, mock.Anything, namespace.ID("namespace-id"), namespace.Name("namespace")).Return(int64(10), nil).Once()
env.OnActivity(a.EnsureNoExecutionsAdvVisibilityActivity, mock.Anything, namespace.ID("namespace-id"), namespace.Name("namespace"), 0).
Return(stderrors.New("specific_error_from_activity")).
Expand Down Expand Up @@ -168,6 +170,7 @@ func Test_ReclaimResourcesWorkflow_EnsureNoExecutionsActivity_ExecutionsStillExi
ErrorCount: 0,
}, nil).Once()

env.OnActivity(la.GetNamespaceCacheRefreshInterval, mock.Anything).Return(10*time.Second, nil).Once()
env.OnActivity(la.CountExecutionsAdvVisibilityActivity, mock.Anything, namespace.ID("namespace-id"), namespace.Name("namespace")).Return(int64(10), nil).Once()
env.OnActivity(a.EnsureNoExecutionsAdvVisibilityActivity, mock.Anything, namespace.ID("namespace-id"), namespace.Name("namespace"), 0).
Return(errors.NewExecutionsStillExist(1)).
Expand Down Expand Up @@ -233,16 +236,17 @@ func Test_ReclaimResourcesWorkflow_NoActivityMocks_Success(t *testing.T) {

a := &Activities{
visibilityManager: visibilityManager,
metricsHandler: metrics.NoopMetricsHandler,
logger: log.NewTestLogger(),
}
la := &LocalActivities{
visibilityManager: visibilityManager,
metadataManager: metadataManager,
metricsHandler: metrics.NoopMetricsHandler,
logger: log.NewTestLogger(),
visibilityManager: visibilityManager,
metadataManager: metadataManager,
namespaceCacheRefreshInterval: dynamicconfig.GetDurationPropertyFn(10 * time.Second),

logger: log.NewTestLogger(),
}

env.RegisterActivity(la.GetNamespaceCacheRefreshInterval)
env.RegisterActivity(la.CountExecutionsAdvVisibilityActivity)
env.RegisterActivity(a.EnsureNoExecutionsAdvVisibilityActivity)
env.RegisterActivity(la.DeleteNamespaceActivity)
Expand Down Expand Up @@ -312,15 +316,15 @@ func Test_ReclaimResourcesWorkflow_NoActivityMocks_NoProgressMade(t *testing.T)

a := &Activities{
visibilityManager: visibilityManager,
metricsHandler: metrics.NoopMetricsHandler,
logger: log.NewTestLogger(),
}
la := &LocalActivities{
visibilityManager: visibilityManager,
metricsHandler: metrics.NoopMetricsHandler,
logger: log.NewTestLogger(),
visibilityManager: visibilityManager,
namespaceCacheRefreshInterval: dynamicconfig.GetDurationPropertyFn(10 * time.Second),
logger: log.NewTestLogger(),
}

env.RegisterActivity(la.GetNamespaceCacheRefreshInterval)
env.RegisterActivity(la.CountExecutionsAdvVisibilityActivity)
env.RegisterActivity(a.EnsureNoExecutionsAdvVisibilityActivity)

Expand Down Expand Up @@ -373,6 +377,7 @@ func Test_ReclaimResourcesWorkflow_UpdateDeleteDelay(t *testing.T) {
ErrorCount: 0,
}, nil).Once()

env.OnActivity(la.GetNamespaceCacheRefreshInterval, mock.Anything).Return(10*time.Second, nil).Once()
env.OnActivity(la.CountExecutionsAdvVisibilityActivity, mock.Anything, namespace.ID("namespace-id"), namespace.Name("namespace")).Return(int64(10), nil).Once()
env.OnActivity(a.EnsureNoExecutionsAdvVisibilityActivity, mock.Anything, namespace.ID("namespace-id"), namespace.Name("namespace"), 0).Return(nil).Once()

Expand Down
Loading