feat: implement distributed lock by gorm

app/dubbo-admin/cmd/run.go

@@ -30,6 +30,7 @@ import (
 	"github.com/apache/dubbo-admin/pkg/config/app"
 	"github.com/apache/dubbo-admin/pkg/core/bootstrap"
 	"github.com/apache/dubbo-admin/pkg/core/logger"
+	_ "github.com/apache/dubbo-admin/pkg/lock/gorm"
 	dubboversion "github.com/apache/dubbo-admin/pkg/version"
 )
 

pkg/common/bizerror/error.go

@@ -46,6 +46,8 @@ const (
 	YamlError       ErrorCode = "YamlError"
 	NotFoundError   ErrorCode = "NotFoundError"
 	NetWorkError    ErrorCode = "NetWorkError"
+	LockNotHeld     ErrorCode = "LockNotHeld"
+	LockExpired     ErrorCode = "LockExpired"
 )
 
 type bizError struct {

pkg/common/constants/lock.go

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package constants
+
+import "time"
+
+const (
+	// DefaultLockTimeout is the default timeout for distributed lock operations
+	// This timeout applies to lock acquisition, renewal, and release operations
+	DefaultLockTimeout = 30 * time.Second
+
+	// DefaultAutoRenewThreshold is the TTL threshold above which auto-renewal is enabled
+	// Locks with TTL longer than this value will be automatically renewed
+	DefaultAutoRenewThreshold = 10 * time.Second
+
+	// DefaultUnlockTimeout is the timeout for unlock operations in deferred cleanup
+	DefaultUnlockTimeout = 5 * time.Second
+
+	// DefaultRenewTimeout is the timeout for lock renewal operations
+	DefaultRenewTimeout = 5 * time.Second
+
+	// DefaultLockRetryInterval is the interval between lock acquisition retry attempts
+	DefaultLockRetryInterval = 100 * time.Millisecond
+
+	// DefaultCleanupInterval is the interval for periodic expired lock cleanup
+	DefaultCleanupInterval = 5 * time.Minute
+
+	// DefaultCleanupTimeout is the timeout for cleanup operations
+	DefaultCleanupTimeout = 30 * time.Second
+)
+
+// Lock key prefixes for different resource types
+const (
+	// TagRouteKeyPrefix is the prefix for tag route lock keys
+	TagRouteKeyPrefix = "tag_route"
+
+	// ConfiguratorRuleKeyPrefix is the prefix for configurator rule lock keys
+	ConfiguratorRuleKeyPrefix = "configurator_rule"
+
+	// ConditionRuleKeyPrefix is the prefix for condition rule lock keys
+	ConditionRuleKeyPrefix = "condition_rule"
+)

pkg/console/context/context.go

@@ -19,6 +19,7 @@ package context
 
 import (
 	ctx "context"
+	"github.com/apache/dubbo-admin/pkg/core/lock"
 
 	"github.com/apache/dubbo-admin/pkg/config/app"
 	"github.com/apache/dubbo-admin/pkg/console/counter"
@@ -33,6 +34,7 @@ type Context interface {
 	Config() app.AdminConfig
 
 	AppContext() ctx.Context
+	LockManager() lock.Lock
 }
 
 var _ Context = &context{}
@@ -71,3 +73,11 @@ func (c *context) CounterManager() counter.CounterManager {
 	}
 	return managerComp.CounterManager()
 }
+
+func (c *context) LockManager() lock.Lock {
+	distributedLock, err := lock.GetLockFromRuntime(c.coreRt)
+	if err != nil {
+		return nil
+	}
+	return distributedLock
+}

pkg/console/service/condition_rule.go

@@ -18,6 +18,8 @@
 package service
 
 import (
+	"github.com/apache/dubbo-admin/pkg/common/constants"
+	"github.com/apache/dubbo-admin/pkg/core/lock"
 	"github.com/duke-git/lancet/v2/slice"
 	"github.com/duke-git/lancet/v2/strutil"
 
@@ -106,6 +108,17 @@ func GetConditionRule(ctx context.Context, name string, mesh string) (*meshresou
 }
 
 func UpdateConditionRule(ctx context.Context, res *meshresource.ConditionRouteResource) error {
+	lockMgr := ctx.LockManager()
+	if lockMgr == nil {
+		return updateConditionRuleUnsafe(ctx, res)
+	}
+	lockKey := lock.BuildConditionRuleLockKey(res.Mesh, res.Name)
+	return lockMgr.WithLock(ctx.AppContext(), lockKey, constants.DefaultLockTimeout, func() error {
+		return updateConditionRuleUnsafe(ctx, res)
+	})
+}
+
+func updateConditionRuleUnsafe(ctx context.Context, res *meshresource.ConditionRouteResource) error {
 	if err := ctx.ResourceManager().Update(res); err != nil {
 		logger.Warnf("update %s condition failed with error: %s", res.Name, err.Error())
 		return err
@@ -114,6 +127,17 @@ func UpdateConditionRule(ctx context.Context, res *meshresource.ConditionRouteRe
 }
 
 func CreateConditionRule(ctx context.Context, res *meshresource.ConditionRouteResource) error {
+	lockMgr := ctx.LockManager()
+	if lockMgr == nil {
+		return createConditionRuleUnsafe(ctx, res)
+	}
+	lockKey := lock.BuildConditionRuleLockKey(res.Mesh, res.Name)
+	return lockMgr.WithLock(ctx.AppContext(), lockKey, constants.DefaultLockTimeout, func() error {
+		return createConditionRuleUnsafe(ctx, res)
+	})
+}
+
+func createConditionRuleUnsafe(ctx context.Context, res *meshresource.ConditionRouteResource) error {
 	if err := ctx.ResourceManager().Add(res); err != nil {
 		logger.Warnf("create %s condition failed with error: %s", res.Name, err.Error())
 		return err
@@ -122,6 +146,17 @@ func CreateConditionRule(ctx context.Context, res *meshresource.ConditionRouteRe
 }
 
 func DeleteConditionRule(ctx context.Context, name string, mesh string) error {
+	lockMgr := ctx.LockManager()
+	if lockMgr == nil {
+		return deleteConditionRuleUnsafe(ctx, name, mesh)
+	}
+	lockKey := lock.BuildConditionRuleLockKey(mesh, name)
+	return lockMgr.WithLock(ctx.AppContext(), lockKey, constants.DefaultLockTimeout, func() error {
+		return deleteConditionRuleUnsafe(ctx, name, mesh)
+	})
+}
+
+func deleteConditionRuleUnsafe(ctx context.Context, name string, mesh string) error {
 	if err := ctx.ResourceManager().DeleteByKey(meshresource.ConditionRouteKind, mesh, coremodel.BuildResourceKey(mesh, name)); err != nil {
 		return err
 	}

pkg/console/service/configurator_rule.go

@@ -18,6 +18,8 @@
 package service
 
 import (
+	"github.com/apache/dubbo-admin/pkg/common/constants"
+	"github.com/apache/dubbo-admin/pkg/core/lock"
 	"github.com/duke-git/lancet/v2/slice"
 
 	"github.com/apache/dubbo-admin/pkg/common/bizerror"
@@ -114,6 +116,17 @@ func GetConfigurator(ctx consolectx.Context, name string, mesh string) (*meshres
 }
 
 func UpdateConfigurator(ctx consolectx.Context, res *meshresource.DynamicConfigResource) error {
+	lockMgr := ctx.LockManager()
+	if lockMgr == nil {
+		return updateConfiguratorUnsafe(ctx, res)
+	}
+	lockKey := lock.BuildConfiguratorRuleLockKey(res.Mesh, res.Name)
+	return lockMgr.WithLock(ctx.AppContext(), lockKey, constants.DefaultLockTimeout, func() error {
+		return updateConfiguratorUnsafe(ctx, res)
+	})
+}
+
+func updateConfiguratorUnsafe(ctx consolectx.Context, res *meshresource.DynamicConfigResource) error {
 	if err := ctx.ResourceManager().Update(res); err != nil {
 		logger.Warnf("update %s configurator failed with error: %s", res.Name, err.Error())
 		return err
@@ -122,6 +135,17 @@ func UpdateConfigurator(ctx consolectx.Context, res *meshresource.DynamicConfigR
 }
 
 func CreateConfigurator(ctx consolectx.Context, res *meshresource.DynamicConfigResource) error {
+	lockMgr := ctx.LockManager()
+	if lockMgr == nil {
+		return createConfiguratorUnsafe(ctx, res)
+	}
+	lockKey := lock.BuildConfiguratorRuleLockKey(res.Mesh, res.Name)
+	return lockMgr.WithLock(ctx.AppContext(), lockKey, constants.DefaultLockTimeout, func() error {
+		return createConfiguratorUnsafe(ctx, res)
+	})
+}
+
+func createConfiguratorUnsafe(ctx consolectx.Context, res *meshresource.DynamicConfigResource) error {
 	if err := ctx.ResourceManager().Add(res); err != nil {
 		logger.Warnf("create %s configurator failed with error: %s", res.Name, err.Error())
 		return err
@@ -130,6 +154,17 @@ func CreateConfigurator(ctx consolectx.Context, res *meshresource.DynamicConfigR
 }
 
 func DeleteConfigurator(ctx consolectx.Context, name string, mesh string) error {
+	lockMgr := ctx.LockManager()
+	if lockMgr == nil {
+		return deleteConfiguratorUnsafe(ctx, name, mesh)
+	}
+	lockKey := lock.BuildConfiguratorRuleLockKey(mesh, name)
+	return lockMgr.WithLock(ctx.AppContext(), lockKey, constants.DefaultLockTimeout, func() error {
+		return deleteConfiguratorUnsafe(ctx, name, mesh)
+	})
+}
+
+func deleteConfiguratorUnsafe(ctx consolectx.Context, name string, mesh string) error {
 	if err := ctx.ResourceManager().DeleteByKey(meshresource.DynamicConfigKind, mesh, coremodel.BuildResourceKey(mesh, name)); err != nil {
 		logger.Warnf("delete %s configurator failed with error: %s", name, err.Error())
 		return err

pkg/console/service/tag_rule.go

@@ -18,6 +18,8 @@
 package service
 
 import (
+	"github.com/apache/dubbo-admin/pkg/common/constants"
+	"github.com/apache/dubbo-admin/pkg/core/lock"
 	"github.com/duke-git/lancet/v2/slice"
 
 	"github.com/apache/dubbo-admin/pkg/common/bizerror"
@@ -112,6 +114,19 @@ func GetTagRule(ctx consolectx.Context, name string, mesh string) (*meshresource
 }
 
 func UpdateTagRule(ctx consolectx.Context, res *meshresource.TagRouteResource) error {
+	lockMgr := ctx.LockManager()
+	if lockMgr == nil {
+		return updateTagRuleUnsafe(ctx, res)
+	}
+
+	lockKey := lock.BuildTagRouteLockKey(res.Mesh, res.Name)
+
+	return lockMgr.WithLock(ctx.AppContext(), lockKey, constants.DefaultLockTimeout, func() error {
+		return updateTagRuleUnsafe(ctx, res)
+	})
+}
+
+func updateTagRuleUnsafe(ctx consolectx.Context, res *meshresource.TagRouteResource) error {
 	err := ctx.ResourceManager().Update(res)
 	if err != nil {
 		logger.Warnf("update tag rule %s error: %v", res.Name, err)
@@ -121,6 +136,19 @@ func UpdateTagRule(ctx consolectx.Context, res *meshresource.TagRouteResource) e
 }
 
 func CreateTagRule(ctx consolectx.Context, res *meshresource.TagRouteResource) error {
+	lockMgr := ctx.LockManager()
+	if lockMgr == nil {
+		return createTagRuleUnsafe(ctx, res)
+	}
+
+	lockKey := lock.BuildTagRouteLockKey(res.Mesh, res.Name)
+
+	return lockMgr.WithLock(ctx.AppContext(), lockKey, constants.DefaultLockTimeout, func() error {
+		return createTagRuleUnsafe(ctx, res)
+	})
+}
+
+func createTagRuleUnsafe(ctx consolectx.Context, res *meshresource.TagRouteResource) error {
 	err := ctx.ResourceManager().Add(res)
 	if err != nil {
 		logger.Warnf("create tag rule %s error: %v", res.Name, err)
@@ -130,6 +158,17 @@ func CreateTagRule(ctx consolectx.Context, res *meshresource.TagRouteResource) e
 }
 
 func DeleteTagRule(ctx consolectx.Context, name string, mesh string) error {
+	lockMgr := ctx.LockManager()
+	if lockMgr == nil {
+		return deleteTagRuleUnsafe(ctx, name, mesh)
+	}
+	lockKey := lock.BuildTagRouteLockKey(mesh, name)
+	return lockMgr.WithLock(ctx.AppContext(), lockKey, constants.DefaultLockTimeout, func() error {
+		return deleteTagRuleUnsafe(ctx, name, mesh)
+	})
+}
+
+func deleteTagRuleUnsafe(ctx consolectx.Context, name string, mesh string) error {
 	err := ctx.ResourceManager().DeleteByKey(meshresource.TagRouteKind, mesh, coremodel.BuildResourceKey(mesh, name))
 	if err != nil {
 		logger.Warnf("delete tag rule %s error: %v", name, err)

pkg/core/bootstrap/bootstrap.go

@@ -24,6 +24,7 @@ import (
 	"github.com/apache/dubbo-admin/pkg/common/bizerror"
 	"github.com/apache/dubbo-admin/pkg/config/app"
 	"github.com/apache/dubbo-admin/pkg/console/counter"
+	"github.com/apache/dubbo-admin/pkg/core/lock"
 	"github.com/apache/dubbo-admin/pkg/core/logger"
 	"github.com/apache/dubbo-admin/pkg/core/runtime"
 	"github.com/apache/dubbo-admin/pkg/diagnostics"
@@ -128,6 +129,7 @@ func (sb *SmartBootstrapper) gatherComponents() ([]runtime.Component, error) {
 	}{
 		{"CounterManager", counter.ComponentType},
 		{"DiagnosticsServer", diagnostics.DiagnosticsServer},
+		{"DistributedLock", lock.DistributedLockComponent},
 	}
 
 	for _, comp := range optionalComps {