Add a system workflow to manage config

common/persistence/cluster_metadata_store.go

@@ -238,8 +238,7 @@ func immutableFieldsChanged(old *persistencespb.ClusterMetadata, cur *persistenc
 		return true
 	}
 	if old.IsGlobalNamespaceEnabled {
-		if (old.FailoverVersionIncrement != 0 && old.FailoverVersionIncrement != cur.FailoverVersionIncrement) ||
-			(old.InitialFailoverVersion != 0 && old.InitialFailoverVersion != cur.InitialFailoverVersion) {
+		if old.InitialFailoverVersion != 0 && old.InitialFailoverVersion != cur.InitialFailoverVersion {
 			return true
 		}
 	}

service/worker/fx.go

@@ -54,6 +54,7 @@ import (
 	"go.temporal.io/server/service/worker/dlq"
 	"go.temporal.io/server/service/worker/migration"
 	"go.temporal.io/server/service/worker/scheduler"
+	"go.temporal.io/server/service/worker/systemconfig"
 	"go.temporal.io/server/service/worker/workerdeployment"
 	"go.uber.org/fx"
 )
@@ -69,6 +70,7 @@ var Module = fx.Options(
 	workerdeployment.Module,
 	dlq.Module,
 	dynamicconfig.Module,
+	systemconfig.Module,
 	fx.Provide(
 		func(c resource.HistoryClient) dlq.HistoryClient {
 			return c

service/worker/systemconfig/activities.go

@@ -0,0 +1,94 @@
+// 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.
+
+package systemconfig
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"go.temporal.io/server/common/persistence"
+)
+
+type (
+	UpdateFailoverVersionIncrementInput struct {
+		CurrentFVI int64
+		NewFVI     int64
+	}
+
+	UpdateFailoverVersionIncrementOutput struct {
+	}
+
+	activities struct {
+		currentClusterName     string
+		clusterMetadataManager persistence.ClusterMetadataManager
+	}
+)
+
+func (a *activities) UpdateFVI(
+	ctx context.Context,
+	input UpdateFailoverVersionIncrementInput,
+) (UpdateFailoverVersionIncrementOutput, error) {
+	request := &persistence.GetClusterMetadataRequest{
+		ClusterName: a.currentClusterName,
+	}
+	if input.NewFVI <= 0 {
+		return UpdateFailoverVersionIncrementOutput{},
+			errors.New("invalid failover version increment")
+	}
+	resp, err := a.clusterMetadataManager.GetClusterMetadata(ctx, request)
+	if err != nil {
+		return UpdateFailoverVersionIncrementOutput{}, err
+	}
+	metadata := resp.ClusterMetadata
+	if metadata.FailoverVersionIncrement != input.CurrentFVI {
+		return UpdateFailoverVersionIncrementOutput{},
+			errors.New(fmt.Sprintf("failover version increment %v does not match input version %v", metadata.FailoverVersionIncrement, input.CurrentFVI))
+	}
+	if metadata.InitialFailoverVersion == input.NewFVI {
+		return UpdateFailoverVersionIncrementOutput{}, nil
+	}
+	if metadata.InitialFailoverVersion > input.NewFVI {
+		return UpdateFailoverVersionIncrementOutput{},
+			errors.New(fmt.Sprintf("failover version increment %v is less than initial failover version %v", input.NewFVI, metadata.InitialFailoverVersion))
+	}
+	if !metadata.IsGlobalNamespaceEnabled {
+		return UpdateFailoverVersionIncrementOutput{},
+			errors.New("please update failover version increment from application yaml file")
+	}
+
+	metadata.FailoverVersionIncrement = input.NewFVI
+	applied, err := a.clusterMetadataManager.SaveClusterMetadata(ctx, &persistence.SaveClusterMetadataRequest{
+		ClusterMetadata: metadata,
+		Version:         resp.Version,
+	})
+	if err != nil {
+		return UpdateFailoverVersionIncrementOutput{}, err
+	}
+	if !applied {
+		return UpdateFailoverVersionIncrementOutput{}, errors.New("new failover version increment did not apply")
+	}
+	return UpdateFailoverVersionIncrementOutput{}, nil
+}

service/worker/systemconfig/fx.go

@@ -0,0 +1,88 @@
+// 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.
+
+package systemconfig
+
+import (
+	sdkworker "go.temporal.io/sdk/worker"
+	"go.temporal.io/sdk/workflow"
+	"go.temporal.io/server/common/cluster"
+	"go.temporal.io/server/common/persistence"
+	workercommon "go.temporal.io/server/service/worker/common"
+	"go.uber.org/fx"
+)
+
+type (
+	initParams struct {
+		fx.In
+		ClusterMetadataManager persistence.ClusterMetadataManager
+		ClusterMetadata        cluster.Metadata
+	}
+
+	fxResult struct {
+		fx.Out
+		Component workercommon.WorkerComponent `group:"workerComponent"`
+	}
+
+	clusterWorkerComponent struct {
+		initParams
+	}
+)
+
+var Module = fx.Options(
+	fx.Provide(NewResult),
+)
+
+func NewResult(params initParams) fxResult {
+	component := &clusterWorkerComponent{
+		initParams: params,
+	}
+	return fxResult{
+		Component: component,
+	}
+}
+
+func (wc *clusterWorkerComponent) RegisterWorkflow(registry sdkworker.Registry) {
+	registry.RegisterWorkflowWithOptions(UpdateFailoverVersionIncrementWorkflow, workflow.RegisterOptions{Name: updateFailoverVersionIncrementWorkflowName})
+}
+
+func (wc *clusterWorkerComponent) DedicatedWorkflowWorkerOptions() *workercommon.DedicatedWorkerOptions {
+	// Use default worker
+	return nil
+}
+
+func (wc *clusterWorkerComponent) RegisterActivities(registry sdkworker.Registry) {
+	registry.RegisterActivity(wc.activities())
+}
+
+func (wc *clusterWorkerComponent) DedicatedActivityWorkerOptions() *workercommon.DedicatedWorkerOptions {
+	return nil
+}
+
+func (wc *clusterWorkerComponent) activities() *activities {
+	return &activities{
+		currentClusterName:     wc.ClusterMetadata.GetCurrentClusterName(),
+		clusterMetadataManager: wc.ClusterMetadataManager,
+	}
+}

service/worker/systemconfig/update_failover_version.go

@@ -0,0 +1,56 @@
+// 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.
+
+package systemconfig
+
+import (
+	"time"
+
+	"go.temporal.io/sdk/workflow"
+)
+
+const (
+	updateFailoverVersionIncrementWorkflowName = "update-failover-version-increment"
+)
+
+type (
+	UpdateFailoverVersionIncrementParams struct {
+		CurrentFVI int64
+		NewFVI     int64
+	}
+)
+
+func UpdateFailoverVersionIncrementWorkflow(
+	ctx workflow.Context,
+	params UpdateFailoverVersionIncrementParams,
+) error {
+	activityOptions := workflow.ActivityOptions{
+		StartToCloseTimeout: 1 * time.Minute,
+	}
+	future := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, activityOptions), "UpdateFVI", UpdateFailoverVersionIncrementInput{
+		CurrentFVI: params.CurrentFVI,
+		NewFVI:     params.NewFVI,
+	})
+	return future.Get(ctx, nil)
+}