Skip to content

Commit 57ec9df

Browse files
justinkasemanjustinkaseman
committed
Remove copied Gateway Ratelimiter. Fill default configs
1 parent 794fc0b commit 57ec9df

File tree

10 files changed

+120
-165
lines changed

10 files changed

+120
-165
lines changed

‎core/capabilities/compute/compute_test.go

+6-7
Original file line numberDiff line numberDiff line change
@@ -22,11 +22,10 @@ import (
2222

2323
corecapabilities "github.com/smartcontractkit/chainlink/v2/core/capabilities"
2424
"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi"
25-
"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi/common"
2625
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/api"
2726
gcmocks "github.com/smartcontractkit/chainlink/v2/core/services/gateway/connector/mocks"
2827
ghcapabilities "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
29-
gwcommon "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
28+
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
3029
)
3130

3231
const (
@@ -38,12 +37,12 @@ const (
3837
var defaultConfig = Config{
3938
ServiceConfig: webapi.ServiceConfig{
4039
OutgoingRateLimiter: common.RateLimiterConfig{
41-
GlobalRPS: 100.0,
42-
GlobalBurst: 100,
43-
PerWorkflowRPS: 100.0,
44-
PerWorkflowBurst: 100,
40+
GlobalRPS: 100.0,
41+
GlobalBurst: 100,
42+
PerSenderRPS: 100.0,
43+
PerSenderBurst: 100,
4544
},
46-
IncomingRateLimiter: gwcommon.RateLimiterConfig{
45+
RateLimiter: common.RateLimiterConfig{
4746
GlobalRPS: 100.0,
4847
GlobalBurst: 100,
4948
PerSenderRPS: 100.0,

‎core/capabilities/webapi/common/ratelimiter.go

-68
This file was deleted.

‎core/capabilities/webapi/common/ratelimiter_test.go

-35
This file was deleted.

‎core/capabilities/webapi/outgoing_connector_handler.go

+48-10
Original file line numberDiff line numberDiff line change
@@ -11,15 +11,21 @@ import (
1111

1212
"github.com/smartcontractkit/chainlink-common/pkg/logger"
1313
"github.com/smartcontractkit/chainlink-common/pkg/services"
14-
common "github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi/common"
1514
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/api"
1615
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/connector"
1716
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
18-
gwcommon "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
17+
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
1918
)
2019

2120
const (
22-
defaultFetchTimeoutMs = 20_000
21+
DefaultGlobalRPS = 100.0
22+
DefaultGlobalBurst = 100
23+
DefaultPerSenderRPS = 100.0
24+
DefaultPerSenderBurst = 100
25+
DefaultWorkflowRPS = 5.0
26+
DefaultWorkflowBurst = 50
27+
defaultFetchTimeoutMs = 20_000
28+
2329
errorOutgoingRatelimitGlobal = "global limit of gateways requests has been exceeded"
2430
errorOutgoingRatelimitWorkflow = "workflow exceeded limit of gateways requests"
2531
errorIncomingRatelimitGlobal = "message from gateway exceeded global rate limit"
@@ -34,18 +40,19 @@ type OutgoingConnectorHandler struct {
3440
gatewaySelector *RoundRobinSelector
3541
method string
3642
lggr logger.Logger
37-
incomingRateLimiter *gwcommon.RateLimiter
43+
incomingRateLimiter *common.RateLimiter
3844
outgoingRateLimiter *common.RateLimiter
3945
responses *responses
4046
}
4147

4248
func NewOutgoingConnectorHandler(gc connector.GatewayConnector, config ServiceConfig, method string, lgger logger.Logger) (*OutgoingConnectorHandler, error) {
43-
outgoingRateLimiter, err := common.NewRateLimiter(config.OutgoingRateLimiter)
49+
outgoingRLCfg := outgoingRateLimiterConfigDefaults(config.OutgoingRateLimiter)
50+
outgoingRateLimiter, err := common.NewRateLimiter(outgoingRLCfg)
4451
if err != nil {
4552
return nil, err
4653
}
47-
48-
incomingRateLimiter, err := gwcommon.NewRateLimiter(config.IncomingRateLimiter)
54+
incomingRLCfg := incomingRateLimiterConfigDefaults(config.RateLimiter)
55+
incomingRateLimiter, err := common.NewRateLimiter(incomingRLCfg)
4956
if err != nil {
5057
return nil, err
5158
}
@@ -70,7 +77,7 @@ func NewOutgoingConnectorHandler(gc connector.GatewayConnector, config ServiceCo
7077
func (c *OutgoingConnectorHandler) HandleSingleNodeRequest(ctx context.Context, messageID string, req capabilities.Request) (*api.Message, error) {
7178
lggr := logger.With(c.lggr, "messageID", messageID, "workflowID", req.WorkflowID)
7279

73-
workflowAllow, globalAllow := c.outgoingRateLimiter.Allow(req.WorkflowID)
80+
workflowAllow, globalAllow := c.outgoingRateLimiter.AllowVerbose(req.WorkflowID)
7481
if !workflowAllow {
7582
return nil, errors.New(errorOutgoingRatelimitWorkflow)
7683
}
@@ -129,7 +136,7 @@ func (c *OutgoingConnectorHandler) HandleSingleNodeRequest(ctx context.Context,
129136
select {
130137
case resp := <-ch:
131138
switch resp.Body.Method {
132-
case api.Method_InternalError:
139+
case api.MethodInternalError:
133140
var errPayload api.JsonRPCError
134141
err := json.Unmarshal(resp.Body.Payload, &errPayload)
135142
if err != nil {
@@ -179,7 +186,7 @@ func (c *OutgoingConnectorHandler) HandleGatewayMessage(ctx context.Context, gat
179186
errMsg := api.Message{
180187
Body: api.MessageBody{
181188
MessageId: body.MessageId,
182-
Method: api.Method_InternalError,
189+
Method: api.MethodInternalError,
183190
Payload: errPayload,
184191
},
185192
}
@@ -228,6 +235,37 @@ func (c *OutgoingConnectorHandler) Name() string {
228235
return c.lggr.Name()
229236
}
230237

238+
func incomingRateLimiterConfigDefaults(config common.RateLimiterConfig) common.RateLimiterConfig {
239+
if config.GlobalBurst == 0 {
240+
config.GlobalBurst = DefaultGlobalBurst
241+
}
242+
if config.GlobalRPS == 0 {
243+
config.GlobalRPS = DefaultGlobalRPS
244+
}
245+
if config.PerSenderBurst == 0 {
246+
config.PerSenderBurst = DefaultPerSenderBurst
247+
}
248+
if config.PerSenderRPS == 0 {
249+
config.PerSenderRPS = DefaultPerSenderRPS
250+
}
251+
return config
252+
}
253+
func outgoingRateLimiterConfigDefaults(config common.RateLimiterConfig) common.RateLimiterConfig {
254+
if config.GlobalBurst == 0 {
255+
config.GlobalBurst = DefaultGlobalBurst
256+
}
257+
if config.GlobalRPS == 0 {
258+
config.GlobalRPS = DefaultGlobalRPS
259+
}
260+
if config.PerSenderBurst == 0 {
261+
config.PerSenderBurst = DefaultWorkflowBurst
262+
}
263+
if config.PerSenderRPS == 0 {
264+
config.PerSenderRPS = DefaultWorkflowRPS
265+
}
266+
return config
267+
}
268+
231269
func validMethod(method string) bool {
232270
switch method {
233271
case capabilities.MethodWebAPITarget, capabilities.MethodComputeAction, capabilities.MethodWorkflowSyncer:

‎core/capabilities/webapi/outgoing_connector_handler_test.go

+35-14
Original file line numberDiff line numberDiff line change
@@ -5,17 +5,17 @@ import (
55
"encoding/json"
66
"testing"
77

8+
"github.com/pelletier/go-toml"
89
"github.com/stretchr/testify/assert"
910
"github.com/stretchr/testify/mock"
1011
"github.com/stretchr/testify/require"
1112

1213
"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
13-
"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi/common"
1414
"github.com/smartcontractkit/chainlink/v2/core/logger"
1515
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/api"
1616
gcmocks "github.com/smartcontractkit/chainlink/v2/core/services/gateway/connector/mocks"
1717
ghcapabilities "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
18-
gwcommon "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
18+
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
1919
"github.com/smartcontractkit/chainlink/v2/core/utils/matches"
2020
)
2121

@@ -199,12 +199,12 @@ func TestHandleSingleNodeRequest(t *testing.T) {
199199
testURL := "http://localhost:8080"
200200
var config = ServiceConfig{
201201
OutgoingRateLimiter: common.RateLimiterConfig{
202-
GlobalRPS: 2.0,
203-
GlobalBurst: 2,
204-
PerWorkflowRPS: 1.0,
205-
PerWorkflowBurst: 1,
202+
GlobalRPS: 2.0,
203+
GlobalBurst: 2,
204+
PerSenderRPS: 1.0,
205+
PerSenderBurst: 1,
206206
},
207-
IncomingRateLimiter: gwcommon.RateLimiterConfig{
207+
RateLimiter: common.RateLimiterConfig{
208208
GlobalRPS: 100.0,
209209
GlobalBurst: 100,
210210
PerSenderRPS: 100.0,
@@ -254,27 +254,27 @@ func TestHandleSingleNodeRequest(t *testing.T) {
254254
WorkflowID: "1",
255255
})
256256
require.Error(t, err)
257-
require.ErrorContains(t, err, error_outgoing_ratelimit_workflow)
257+
require.ErrorContains(t, err, errorOutgoingRatelimitWorkflow)
258258

259259
// Third request should error from global ratelimit
260260
_, err = connectorHandler.HandleSingleNodeRequest(ctx, msgID, ghcapabilities.Request{
261261
URL: testURL,
262262
WorkflowID: "2",
263263
})
264264
require.Error(t, err)
265-
require.ErrorContains(t, err, error_outgoing_ratelimit_global)
265+
require.ErrorContains(t, err, errorOutgoingRatelimitGlobal)
266266
})
267267
}
268268

269269
func newFunctionWithDefaultConfig(t *testing.T, mockFn func(*gcmocks.GatewayConnector)) (*gcmocks.GatewayConnector, *OutgoingConnectorHandler) {
270270
var defaultConfig = ServiceConfig{
271271
OutgoingRateLimiter: common.RateLimiterConfig{
272-
GlobalRPS: 100.0,
273-
GlobalBurst: 100,
274-
PerWorkflowRPS: 100.0,
275-
PerWorkflowBurst: 100,
272+
GlobalRPS: 100.0,
273+
GlobalBurst: 100,
274+
PerSenderRPS: 100.0,
275+
PerSenderBurst: 100,
276276
},
277-
IncomingRateLimiter: gwcommon.RateLimiterConfig{
277+
RateLimiter: common.RateLimiterConfig{
278278
GlobalRPS: 100.0,
279279
GlobalBurst: 100,
280280
PerSenderRPS: 100.0,
@@ -313,3 +313,24 @@ func gatewayResponse(t *testing.T, msgID string) *api.Message {
313313
},
314314
}
315315
}
316+
317+
func TestServiceConfigDefaults(t *testing.T) {
318+
t.Run("fills default RateLimiterConfigs", func(t *testing.T) {
319+
var cfg ServiceConfig
320+
321+
tomlErr := toml.Unmarshal([]byte{}, &cfg)
322+
require.NoError(t, tomlErr)
323+
324+
iRLConf := incomingRateLimiterConfigDefaults(cfg.RateLimiter)
325+
require.Equal(t, DefaultGlobalBurst, iRLConf.GlobalBurst)
326+
require.InDelta(t, DefaultGlobalRPS, iRLConf.GlobalRPS, 0.001)
327+
require.Equal(t, DefaultPerSenderBurst, iRLConf.PerSenderBurst)
328+
require.InDelta(t, DefaultPerSenderRPS, iRLConf.PerSenderRPS, 0.001)
329+
330+
oRLConf := outgoingRateLimiterConfigDefaults(cfg.RateLimiter)
331+
require.Equal(t, DefaultGlobalBurst, oRLConf.GlobalBurst)
332+
require.InDelta(t, DefaultGlobalRPS, oRLConf.GlobalRPS, 0.001)
333+
require.Equal(t, DefaultWorkflowBurst, oRLConf.PerSenderBurst)
334+
require.InDelta(t, DefaultWorkflowRPS, oRLConf.PerSenderRPS, 0.001)
335+
})
336+
}

‎core/capabilities/webapi/target/target_test.go

+6-7
Original file line numberDiff line numberDiff line change
@@ -15,12 +15,11 @@ import (
1515
registrymock "github.com/smartcontractkit/chainlink-common/pkg/types/core/mocks"
1616
"github.com/smartcontractkit/chainlink-common/pkg/values"
1717
"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi"
18-
common "github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi/common"
1918
"github.com/smartcontractkit/chainlink/v2/core/internal/testutils"
2019
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/api"
2120
gcmocks "github.com/smartcontractkit/chainlink/v2/core/services/gateway/connector/mocks"
2221
ghcapabilities "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
23-
gwcommon "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
22+
"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common"
2423
)
2524

2625
const (
@@ -32,12 +31,12 @@ const (
3231

3332
var defaultConfig = webapi.ServiceConfig{
3433
OutgoingRateLimiter: common.RateLimiterConfig{
35-
GlobalRPS: 100.0,
36-
GlobalBurst: 100,
37-
PerWorkflowRPS: 100.0,
38-
PerWorkflowBurst: 100,
34+
GlobalRPS: 100.0,
35+
GlobalBurst: 100,
36+
PerSenderRPS: 100.0,
37+
PerSenderBurst: 100,
3938
},
40-
IncomingRateLimiter: gwcommon.RateLimiterConfig{
39+
RateLimiter: common.RateLimiterConfig{
4140
GlobalRPS: 100.0,
4241
GlobalBurst: 100,
4342
PerSenderRPS: 100.0,

0 commit comments

Comments
 (0)