Remove copied Gateway Ratelimiter. Fill default configs

Author: justinkaseman

core/capabilities/compute/compute_test.go

@@ -22,11 +22,10 @@ import (
 
 	corecapabilities "github.com/smartcontractkit/chainlink/v2/core/capabilities"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi"
-	"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi/common"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/api"
 	gcmocks "github.com/smartcontractkit/chainlink/v2/core/services/gateway/connector/mocks"
 	ghcapabilities "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
-	gwcommon "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
+	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
 )
 
 const (
@@ -38,12 +37,12 @@ const (
 var defaultConfig = Config{
 	ServiceConfig: webapi.ServiceConfig{
 		OutgoingRateLimiter: common.RateLimiterConfig{
-			GlobalRPS:        100.0,
-			GlobalBurst:      100,
-			PerWorkflowRPS:   100.0,
-			PerWorkflowBurst: 100,
+			GlobalRPS:      100.0,
+			GlobalBurst:    100,
+			PerSenderRPS:   100.0,
+			PerSenderBurst: 100,
 		},
-		IncomingRateLimiter: gwcommon.RateLimiterConfig{
+		RateLimiter: common.RateLimiterConfig{
 			GlobalRPS:      100.0,
 			GlobalBurst:    100,
 			PerSenderRPS:   100.0,

core/capabilities/webapi/common/ratelimiter.go

@@ -11,15 +11,21 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
-	common "github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi/common"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/api"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/connector"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
-	gwcommon "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
+	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
 )
 
 const (
-	defaultFetchTimeoutMs          = 20_000
+	DefaultGlobalRPS      = 100.0
+	DefaultGlobalBurst    = 100
+	DefaultPerSenderRPS   = 100.0
+	DefaultPerSenderBurst = 100
+	DefaultWorkflowRPS    = 5.0
+	DefaultWorkflowBurst  = 50
+	defaultFetchTimeoutMs = 20_000
+
 	errorOutgoingRatelimitGlobal   = "global limit of gateways requests has been exceeded"
 	errorOutgoingRatelimitWorkflow = "workflow exceeded limit of gateways requests"
 	errorIncomingRatelimitGlobal   = "message from gateway exceeded global rate limit"
@@ -34,18 +40,19 @@ type OutgoingConnectorHandler struct {
 	gatewaySelector     *RoundRobinSelector
 	method              string
 	lggr                logger.Logger
-	incomingRateLimiter *gwcommon.RateLimiter
+	incomingRateLimiter *common.RateLimiter
 	outgoingRateLimiter *common.RateLimiter
 	responses           *responses
 }
 
 func NewOutgoingConnectorHandler(gc connector.GatewayConnector, config ServiceConfig, method string, lgger logger.Logger) (*OutgoingConnectorHandler, error) {
-	outgoingRateLimiter, err := common.NewRateLimiter(config.OutgoingRateLimiter)
+	outgoingRLCfg := outgoingRateLimiterConfigDefaults(config.OutgoingRateLimiter)
+	outgoingRateLimiter, err := common.NewRateLimiter(outgoingRLCfg)
 	if err != nil {
 		return nil, err
 	}
-
-	incomingRateLimiter, err := gwcommon.NewRateLimiter(config.IncomingRateLimiter)
+	incomingRLCfg := incomingRateLimiterConfigDefaults(config.RateLimiter)
+	incomingRateLimiter, err := common.NewRateLimiter(incomingRLCfg)
 	if err != nil {
 		return nil, err
 	}
@@ -70,7 +77,7 @@ func NewOutgoingConnectorHandler(gc connector.GatewayConnector, config ServiceCo
 func (c *OutgoingConnectorHandler) HandleSingleNodeRequest(ctx context.Context, messageID string, req capabilities.Request) (*api.Message, error) {
 	lggr := logger.With(c.lggr, "messageID", messageID, "workflowID", req.WorkflowID)
 
-	workflowAllow, globalAllow := c.outgoingRateLimiter.Allow(req.WorkflowID)
+	workflowAllow, globalAllow := c.outgoingRateLimiter.AllowVerbose(req.WorkflowID)
 	if !workflowAllow {
 		return nil, errors.New(errorOutgoingRatelimitWorkflow)
 	}
@@ -228,6 +235,37 @@ func (c *OutgoingConnectorHandler) Name() string {
 	return c.lggr.Name()
 }
 
+func incomingRateLimiterConfigDefaults(config common.RateLimiterConfig) common.RateLimiterConfig {
+	if config.GlobalBurst == 0 {
+		config.GlobalBurst = DefaultGlobalBurst
+	}
+	if config.GlobalRPS == 0 {
+		config.GlobalRPS = DefaultGlobalRPS
+	}
+	if config.PerSenderBurst == 0 {
+		config.PerSenderBurst = DefaultPerSenderBurst
+	}
+	if config.PerSenderRPS == 0 {
+		config.PerSenderRPS = DefaultPerSenderRPS
+	}
+	return config
+}
+func outgoingRateLimiterConfigDefaults(config common.RateLimiterConfig) common.RateLimiterConfig {
+	if config.GlobalBurst == 0 {
+		config.GlobalBurst = DefaultGlobalBurst
+	}
+	if config.GlobalRPS == 0 {
+		config.GlobalRPS = DefaultGlobalRPS
+	}
+	if config.PerSenderBurst == 0 {
+		config.PerSenderBurst = DefaultWorkflowBurst
+	}
+	if config.PerSenderRPS == 0 {
+		config.PerSenderRPS = DefaultWorkflowRPS
+	}
+	return config
+}
+
 func validMethod(method string) bool {
 	switch method {
 	case capabilities.MethodWebAPITarget, capabilities.MethodComputeAction, capabilities.MethodWorkflowSyncer:

core/capabilities/webapi/common/ratelimiter_test.go

@@ -5,17 +5,17 @@ import (
 	"encoding/json"
 	"testing"
 
+	"github.com/pelletier/go-toml"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
-	"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi/common"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/api"
 	gcmocks "github.com/smartcontractkit/chainlink/v2/core/services/gateway/connector/mocks"
 	ghcapabilities "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
-	gwcommon "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
+	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
 	"github.com/smartcontractkit/chainlink/v2/core/utils/matches"
 )
 
@@ -199,12 +199,12 @@ func TestHandleSingleNodeRequest(t *testing.T) {
 		testURL := "http://localhost:8080"
 		var config = ServiceConfig{
 			OutgoingRateLimiter: common.RateLimiterConfig{
-				GlobalRPS:        2.0,
-				GlobalBurst:      2,
-				PerWorkflowRPS:   1.0,
-				PerWorkflowBurst: 1,
+				GlobalRPS:      2.0,
+				GlobalBurst:    2,
+				PerSenderRPS:   1.0,
+				PerSenderBurst: 1,
 			},
-			IncomingRateLimiter: gwcommon.RateLimiterConfig{
+			RateLimiter: common.RateLimiterConfig{
 				GlobalRPS:      100.0,
 				GlobalBurst:    100,
 				PerSenderRPS:   100.0,
@@ -254,27 +254,27 @@ func TestHandleSingleNodeRequest(t *testing.T) {
 			WorkflowID: "1",
 		})
 		require.Error(t, err)
-		require.ErrorContains(t, err, error_outgoing_ratelimit_workflow)
+		require.ErrorContains(t, err, errorOutgoingRatelimitWorkflow)
 
 		// Third request should error from global ratelimit
 		_, err = connectorHandler.HandleSingleNodeRequest(ctx, msgID, ghcapabilities.Request{
 			URL:        testURL,
 			WorkflowID: "2",
 		})
 		require.Error(t, err)
-		require.ErrorContains(t, err, error_outgoing_ratelimit_global)
+		require.ErrorContains(t, err, errorOutgoingRatelimitGlobal)
 	})
 }
 
 func newFunctionWithDefaultConfig(t *testing.T, mockFn func(*gcmocks.GatewayConnector)) (*gcmocks.GatewayConnector, *OutgoingConnectorHandler) {
 	var defaultConfig = ServiceConfig{
 		OutgoingRateLimiter: common.RateLimiterConfig{
-			GlobalRPS:        100.0,
-			GlobalBurst:      100,
-			PerWorkflowRPS:   100.0,
-			PerWorkflowBurst: 100,
+			GlobalRPS:      100.0,
+			GlobalBurst:    100,
+			PerSenderRPS:   100.0,
+			PerSenderBurst: 100,
 		},
-		IncomingRateLimiter: gwcommon.RateLimiterConfig{
+		RateLimiter: common.RateLimiterConfig{
 			GlobalRPS:      100.0,
 			GlobalBurst:    100,
 			PerSenderRPS:   100.0,
@@ -313,3 +313,24 @@ func gatewayResponse(t *testing.T, msgID string) *api.Message {
 		},
 	}
 }
+
+func TestServiceConfigDefaults(t *testing.T) {
+	t.Run("fills default RateLimiterConfigs", func(t *testing.T) {
+		var cfg ServiceConfig
+
+		tomlErr := toml.Unmarshal([]byte{}, &cfg)
+		require.NoError(t, tomlErr)
+
+		iRLConf := incomingRateLimiterConfigDefaults(cfg.RateLimiter)
+		require.Equal(t, DefaultGlobalBurst, iRLConf.GlobalBurst)
+		require.InDelta(t, DefaultGlobalRPS, iRLConf.GlobalRPS, 0.001)
+		require.Equal(t, DefaultPerSenderBurst, iRLConf.PerSenderBurst)
+		require.InDelta(t, DefaultPerSenderRPS, iRLConf.PerSenderRPS, 0.001)
+
+		oRLConf := outgoingRateLimiterConfigDefaults(cfg.RateLimiter)
+		require.Equal(t, DefaultGlobalBurst, oRLConf.GlobalBurst)
+		require.InDelta(t, DefaultGlobalRPS, oRLConf.GlobalRPS, 0.001)
+		require.Equal(t, DefaultWorkflowBurst, oRLConf.PerSenderBurst)
+		require.InDelta(t, DefaultWorkflowRPS, oRLConf.PerSenderRPS, 0.001)
+	})
+}

core/capabilities/webapi/outgoing_connector_handler.go

@@ -15,12 +15,11 @@ import (
 	registrymock "github.com/smartcontractkit/chainlink-common/pkg/types/core/mocks"
 	"github.com/smartcontractkit/chainlink-common/pkg/values"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi"
-	common "github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi/common"
 	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/api"
 	gcmocks "github.com/smartcontractkit/chainlink/v2/core/services/gateway/connector/mocks"
 	ghcapabilities "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
-	gwcommon "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
+	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
 )
 
 const (
@@ -32,12 +31,12 @@ const (
 
 var defaultConfig = webapi.ServiceConfig{
 	OutgoingRateLimiter: common.RateLimiterConfig{
-		GlobalRPS:        100.0,
-		GlobalBurst:      100,
-		PerWorkflowRPS:   100.0,
-		PerWorkflowBurst: 100,
+		GlobalRPS:      100.0,
+		GlobalBurst:    100,
+		PerSenderRPS:   100.0,
+		PerSenderBurst: 100,
 	},
-	IncomingRateLimiter: gwcommon.RateLimiterConfig{
+	RateLimiter: common.RateLimiterConfig{
 		GlobalRPS:      100.0,
 		GlobalBurst:    100,
 		PerSenderRPS:   100.0,

core/capabilities/webapi/outgoing_connector_handler_test.go

@@ -1,8 +1,7 @@
 package webapi
 
 import (
-	"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi/common"
-	gwcommon "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
+	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
 )
 
 const (
@@ -16,6 +15,10 @@ const (
 // Note that workflow executions have their own internal timeouts and retries set by the user
 // that are separate from this configuration
 type ServiceConfig struct {
-	IncomingRateLimiter gwcommon.RateLimiterConfig `toml:"incomingRateLimiter" json:"incomingRateLimiter" yaml:"incomingRateLimiter" mapstructure:"incomingRateLimiter"`
-	OutgoingRateLimiter common.RateLimiterConfig   `toml:"outgoingRateLimiter" json:"outgoingRateLimiter" yaml:"outgoingRateLimiter" mapstructure:"outgoingRateLimiter"`
+	// RateLimiter configuration for messages incoming to this node from the gateway.
+	// The sender is a Gateway node, which is identified by the Gateway ID.
+	RateLimiter common.RateLimiterConfig `toml:"incomingRateLimiter" json:"incomingRateLimiter" yaml:"incomingRateLimiter" mapstructure:"incomingRateLimiter"`
+	// RateLimiter configuration for outgoing messages from this node to the gateway.
+	// The sender is a workflow, which is identified by the Workflow ID.
+	OutgoingRateLimiter common.RateLimiterConfig `toml:"outgoingRateLimiter" json:"outgoingRateLimiter" yaml:"outgoingRateLimiter" mapstructure:"outgoingRateLimiter"`
 }