Refactor: Use generalized LRU cache for rate limiter cache.

charts/vela-workflow/templates/definitions/request.yaml

@@ -13,56 +13,72 @@ spec:
   schematic:
     cue:
       template: |
-        import "vela/op"
-        import "vela/http"
-        import "encoding/json"
+        import (
+          "vela/op"
+          "vela/http"
+          "encoding/json"
+        )
 
         req: http.#HTTPDo & {
-        	$params: {
-        		method: parameter.method
-        		url:    parameter.url
-        		request: {
-        			if parameter.body != _|_ {
-        				body: json.Marshal(parameter.body)
-        			}
-        			if parameter.header != _|_ {
-        				header: parameter.header
-        			}
-        			if parameter.headersFromSecret != _|_ {
-        				headersFromSecret: parameter.headersFromSecret
-        			}
-        		}
-        	}
+          $params: {
+            method: parameter.method
+            url:    parameter.url
+            request: {
+              if parameter.body != _|_ {
+                body: json.Marshal(parameter.body)
+              }
+              if parameter.header != _|_ {
+                header: parameter.header
+              }
+              if parameter.timeout != _|_ {
+                timeout: parameter.timeout
+              }
+              if parameter.headersFromSecret != _|_ {
+                headersFromSecret: parameter.headersFromSecret
+              }
+              if parameter.ratelimiter != _|_ {
+                ratelimiter: parameter.ratelimiter
+              }
+            }
+          }
         }
 
         wait: op.#ConditionalWait & {
-        	continue: req.$returns != _|_
-        	message?: "Waiting for response from \(parameter.url)"
+          continue: req.$returns != _|_
+          message?: "Waiting for response from \(parameter.url)"
         }
 
         fail: op.#Steps & {
-        	if req.$returns.statusCode > 400 {
-        		requestFail: op.#Fail & {
-        			message: "request of \(parameter.url) is fail: \(req.$returns.statusCode)"
-        		}
-        	}
+          if req.$returns.statusCode > 400 {
+            requestFail: op.#Fail & {
+              message: "request of \(parameter.url) is fail: \(req.$returns.statusCode)"
+            }
+          }
         }
 
         response: json.Unmarshal(req.$returns.body)
 
         parameter: {
-        	url:    string
-        	method: *"GET" | "POST" | "PUT" | "DELETE"
-        	body?: {...}
-        	header?: [string]: string
-        	// +usage=Headers whose values are sourced from Kubernetes Secrets
-        	headersFromSecret?: [...{
-        		// +usage=The HTTP header name to set
-        		header: string
-        		// +usage=The name of the Kubernetes Secret
-        		secret: string
-        		// +usage=The key within Secret.Data whose value becomes the header value
-        		key: string
-        	}]
+          url:    string
+          method: *"GET" | "POST" | "PUT" | "DELETE"
+          body?: {...}
+          header?: [string]: string
+          // +usage=The timeout of this request (Go duration string, e.g. "30s", "2m", "500ms"). Defaults to 3s when omitted. Invalid values fail when the step runs.
+          timeout?: string & =~"^(0|(([0-9]+(\\.[0-9]*)?|\\.[0-9]+)(ns|us|µs|μs|ms|s|m|h))+)$"
+          // +usage=Headers whose values are sourced from Kubernetes Secrets
+          headersFromSecret?: [...{
+            // +usage=The HTTP header name to set
+            header: string
+            // +usage=The name of the Kubernetes Secret
+            secret: string
+            // +usage=The key within Secret.Data whose value becomes the header value
+            key: string
+          }]
+          // +usage=The rate limiter of the request
+          ratelimiter?: {
+            // +usage=The maximum number of requests allowed within the period. Must be greater than 0.
+            limit: int & >0
+            // +usage=The time window for rate limiting (Go duration string, e.g. "1s", "100ms").
+            period: string & =~"^(0|(([0-9]+(\\.[0-9]*)?|\\.[0-9]+)(ns|us|µs|μs|ms|s|m|h))+)$"
+          }
         }
-

cmd/main.go

@@ -60,6 +60,8 @@ import (
 	"github.com/kubevela/workflow/pkg/features"
 	"github.com/kubevela/workflow/pkg/monitor/watcher"
 	"github.com/kubevela/workflow/pkg/providers"
+	httpProvider "github.com/kubevela/workflow/pkg/providers/http"
+	legacyhttpProvider "github.com/kubevela/workflow/pkg/providers/legacy/http"
 	"github.com/kubevela/workflow/pkg/types"
 	"github.com/kubevela/workflow/pkg/utils"
 	"github.com/kubevela/workflow/pkg/utils/httpguard"
@@ -150,6 +152,9 @@ func main() {
 	flag.CommandLine.AddGoFlagSet(goflag.CommandLine)
 	flag.Parse()
 	wfupgrade.InitCompatibilityCache(context.Background(), wfupgrade.CompatibilityCacheSize)
+	rootCtx := ctrl.SetupSignalHandler()
+	httpProvider.InitRateLimiter(rootCtx)
+	legacyhttpProvider.InitRateLimiter(rootCtx)
 	if logDebug {
 		_ = flag.Set("v", strconv.Itoa(int(common.LogDebug)))
 	}
@@ -324,7 +329,7 @@ func main() {
 	watcher.StartWorkflowRunMetricsWatcher(informer)
 
 	klog.Info("starting manager")
-	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
+	if err := mgr.Start(rootCtx); err != nil {
 		klog.Error(err, "problem running manager")
 		os.Exit(1)
 	}

go.mod

@@ -8,11 +8,10 @@ require (
 	github.com/aliyun/aliyun-log-go-sdk v0.1.38
 	github.com/crossplane/crossplane-runtime v1.16.0
 	github.com/evanphx/json-patch v5.7.0+incompatible
-	github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8
 	github.com/google/go-cmp v0.7.0
 	github.com/hashicorp/go-version v1.6.0
 	github.com/kubevela/kube-trigger v0.1.1-0.20250711201929-51c837aa9bd2
-	github.com/kubevela/pkg v1.11.1-0.20260722232534-455842713731
+	github.com/kubevela/pkg v1.11.1-0.20260826024906-a4214d8d3c39
 	github.com/onsi/ginkgo/v2 v2.23.3
 	github.com/onsi/gomega v1.36.2
 	github.com/pkg/errors v0.9.1
@@ -70,6 +69,7 @@ require (
 	github.com/go-stack/stack v1.8.1 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/cel-go v0.20.1 // indirect
@@ -79,6 +79,7 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/imdario/mergo v0.3.16 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jellydator/ttlcache/v3 v3.0.1 // indirect

go.sum

@@ -266,6 +266,8 @@ github.com/hashicorp/go-version v1.6.0 h1:feTTfFNnjP967rlCxM/I9g701jU+RN74YKx2mO
 github.com/hashicorp/go-version v1.6.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
+github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
 github.com/hashicorp/mdns v1.0.1/go.mod h1:4gW7WsVCke5TE7EPeYliwHlRUyBtfCwuFwuMg2DmyNY=
 github.com/hashicorp/memberlist v0.2.2/go.mod h1:MS2lj3INKhZjWNqd3N0m3J+Jxf3DAOnAH9VT3Sh9MUE=
@@ -320,8 +322,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kubevela/kube-trigger v0.1.1-0.20250711201929-51c837aa9bd2 h1:3FvzYw6qJiAu7OLbOK7PuhDLowtKs3Ugw1gQ2E8jz/Y=
 github.com/kubevela/kube-trigger v0.1.1-0.20250711201929-51c837aa9bd2/go.mod h1:lGeLpXHVxYPulYtVgGatFHFolkmzSmrdyhIQ7BHHJHs=
-github.com/kubevela/pkg v1.11.1-0.20260722232534-455842713731 h1:P9sSZ2YJtyDzDcu28uJ3d0HliiUDjl2VnEZY5Czmo6I=
-github.com/kubevela/pkg v1.11.1-0.20260722232534-455842713731/go.mod h1:EmM4VIyU7KxDmPBq9hG4GpSZbGwiM76/W/8paLBk8wY=
+github.com/kubevela/pkg v1.11.1-0.20260826024906-a4214d8d3c39 h1:GRG4cDE3ALeIBeyzTQwZKeuGXvUjCd6ulWK5W3C6r60=
+github.com/kubevela/pkg v1.11.1-0.20260826024906-a4214d8d3c39/go.mod h1:MZsc/xbV4rZjE2fMWX8+JovRK/zVUb1HfefGNPYrgq8=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=

pkg/providers/http/http.cue

@@ -21,8 +21,8 @@
 			trailer?: [string]: string
 			// +usage=The rate limiter of the request
 			ratelimiter?: {
-				limit:  int
-				period: string
+				limit:  int & >0
+				period: string & =~"^(0|(([0-9]+(\\.[0-9]*)?|\\.[0-9]+)(ns|us|µs|μs|ms|s|m|h))+)$"
 			}
 			// +usage=Headers whose values are resolved from Kubernetes Secrets
 			headersFromSecret?: [...{

pkg/providers/http/http.go

@@ -27,10 +27,12 @@ import (
 	"net/http"
 	neturl "net/url"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/pkg/errors"
 	v1 "k8s.io/api/core/v1"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -49,13 +51,10 @@ const (
 )
 
 var (
-	rateLimiter *ratelimiter.RateLimiter
+	rateLimiterOnce sync.Once
+	rateLimiter     *ratelimiter.RateLimiter
 )
 
-func init() {
-	rateLimiter = ratelimiter.NewRateLimiter(128)
-}
-
 // Request .
 type Request struct {
 	Timeout           string            `json:"timeout,omitempty"`
@@ -120,10 +119,20 @@ func requestPolicy() httpguard.Policy {
 	return policy
 }
 
+// InitRateLimiter initializes the rate limiter. Must be called once at startup.
+func InitRateLimiter(ctx context.Context) {
+	rateLimiterOnce.Do(func() {
+		var err error
+		rateLimiter, err = ratelimiter.NewRateLimiter(ctx, 128)
+		utilruntime.Must(err)
+	})
+}
+
 func runHTTP(ctx context.Context, params *DoParams) (*DoReturns, error) {
 	if utilfeature.DefaultMutableFeatureGate.Enabled(features.DisableWorkflowHTTP) {
 		return nil, errors.New("workflow outbound HTTP is disabled by DisableWorkflowHTTP feature gate")
 	}
+	InitRateLimiter(context.WithoutCancel(ctx))
 	var (
 		err             error
 		header, trailer http.Header

pkg/providers/http/http_test.go

@@ -161,7 +161,9 @@ func TestHttpDo(t *testing.T) {
 	}
 
 	// test ratelimiter
-	rateLimiter = ratelimiter.NewRateLimiter(1)
+	var err error
+	rateLimiter, err = ratelimiter.NewRateLimiter(ctx, 1)
+	require.NoError(t, err)
 	limiterTestCases := []struct {
 		request     RequestVars
 		expectedErr string

pkg/providers/legacy/http/http.cue

@@ -20,8 +20,8 @@
 		trailer?: [string]: string
 		// +usage=The rate limiter of the request
 		ratelimiter?: {
-			limit:  int
-			period: string
+			limit:  int & >0
+			period: string & =~"^(0|(([0-9]+(\\.[0-9]*)?|\\.[0-9]+)(ns|us|µs|μs|ms|s|m|h))+)$"
 		}
 		...
 	}