Add terraform state lock

Author: Elbandi

modules/globallock/globallock.go

@@ -45,6 +45,10 @@ func TryLock(ctx context.Context, key string) (bool, ReleaseFunc, error) {
 	return DefaultLocker().TryLock(ctx, key)
 }
 
+func Unlock(ctx context.Context, key string) error {
+	return DefaultLocker().Unlock(ctx, key)
+}
+
 // LockAndDo tries to acquire a lock for the given key and then calls the given function.
 // It uses the default locker, and it will return an error if failed to acquire the lock.
 func LockAndDo(ctx context.Context, key string, f func(context.Context) error) error {

modules/globallock/locker.go

@@ -32,6 +32,8 @@ type Locker interface {
 	// And if it fails to acquire the lock because it's already locked, not other reasons like redis is down,
 	// it will return false without any error.
 	TryLock(ctx context.Context, key string) (bool, ReleaseFunc, error)
+
+	Unlock(ctx context.Context, key string) error
 }
 
 // ReleaseFunc is a function that releases a lock.

modules/globallock/memory_locker.go

@@ -61,6 +61,11 @@ func (l *memoryLocker) TryLock(_ context.Context, key string) (bool, ReleaseFunc
 	return false, func() {}, nil
 }
 
+func (l *memoryLocker) Unlock(_ context.Context, key string) error {
+	l.locks.Delete(key)
+	return nil
+}
+
 func (l *memoryLocker) tryLock(key string) bool {
 	_, loaded := l.locks.LoadOrStore(key, struct{}{})
 	return !loaded

modules/globallock/redis_locker.go

@@ -64,6 +64,19 @@ func (l *redisLocker) TryLock(ctx context.Context, key string) (bool, ReleaseFun
 	return err == nil, f, err
 }
 
+func (l *redisLocker) Unlock(ctx context.Context, key string) error {
+	mutex, ok := l.mutexM.Load(key)
+	if ok {
+		l.mutexM.Delete(key)
+
+		// It's safe to ignore the error here,
+		// if it failed to unlock, it will be released automatically after the lock expires.
+		// Do not call mutex.UnlockContext(ctx) here, or it will fail to release when ctx has timed out.
+		_, _ = mutex.(*redsync.Mutex).Unlock()
+	}
+	return nil
+}
+
 // Close closes the locker.
 // It will stop extending the locks and refuse to acquire new locks.
 // In actual use, it is not necessary to call this function.

routers/api/packages/api.go

@@ -672,6 +672,10 @@ func CommonRoutes() *web.Router {
 						r.Put("", terraform.UploadPackage)
 						r.Delete("", terraform.DeletePackageFile)
 					}, reqPackageAccess(perm.AccessModeWrite))
+					r.Group("/lock", func() {
+						r.Post("", terraform.LockPackage)
+						r.Delete("", terraform.UnlockPackage)
+					}, reqPackageAccess(perm.AccessModeWrite))
 				})
 			})
 		}, reqPackageAccess(perm.AccessModeRead))

routers/api/packages/terraform/terraform.go

@@ -208,3 +208,49 @@ func DeletePackageFile(ctx *context.Context) {
 
 	ctx.Status(http.StatusNoContent)
 }
+
+// LockPackage locks the specific terraform state.
+func LockPackage(ctx *context.Context) {
+	packageName := ctx.PathParam("packagename")
+	pv, err := packages_model.GetVersionByNameAndVersion(ctx, ctx.Package.Owner.ID, packages_model.TypeTerraform, packageName, ctx.PathParam("filename"))
+	if err != nil {
+		if errors.Is(err, packages_model.ErrPackageNotExist) || errors.Is(err, packages_model.ErrPackageFileNotExist) {
+			apiError(ctx, http.StatusNotFound, err)
+			return
+		}
+		apiError(ctx, http.StatusInternalServerError, err)
+		return
+	}
+
+	//log.Error("pv: %+v", pv)
+	ok, _, err := globallock.TryLock(ctx, fmt.Sprintf("%s/%s", packageName, pv.LowerVersion))
+	if err != nil {
+		apiError(ctx, http.StatusInternalServerError, err)
+		return
+	}
+	if !ok {
+		apiError(ctx, http.StatusLocked, err)
+		return
+	}
+
+	ctx.Status(http.StatusOK)
+}
+
+// UnlockPackage unlock the specific terraform state.
+func UnlockPackage(ctx *context.Context) {
+	packageName := ctx.PathParam("packagename")
+	pv, err := packages_model.GetVersionByNameAndVersion(ctx, ctx.Package.Owner.ID, packages_model.TypeTerraform, packageName, ctx.PathParam("filename"))
+	if err != nil {
+		if errors.Is(err, packages_model.ErrPackageNotExist) || errors.Is(err, packages_model.ErrPackageFileNotExist) {
+			apiError(ctx, http.StatusNotFound, err)
+			return
+		}
+		apiError(ctx, http.StatusInternalServerError, err)
+		return
+	}
+
+	//log.Error("pv: %+v", pv)
+	_ = globallock.Unlock(ctx, fmt.Sprintf("%s/%s", packageName, pv.LowerVersion))
+
+	ctx.Status(http.StatusOK)
+}