From 45dee234c8e62f25cc43da6bb0794a7bb5b76570 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Wed, 28 May 2025 12:31:04 +0300 Subject: [PATCH] fixes after review + new tests added --- .../TenantProfileControllerTest.java | 38 +++++++ .../server/common/data/limit/LimitedApi.java | 8 +- ...mitedApiEntry.java => RateLimitEntry.java} | 6 +- ...LimitedApiUtil.java => RateLimitUtil.java} | 18 ++-- .../DefaultTenantProfileConfiguration.java | 56 +++++----- .../common/data/limit/LimitedApiTest.java | 102 ++++++++++++++++++ ...piUtilTest.java => RateLimitUtilTest.java} | 16 +-- .../server/common/msg/tools/TbRateLimits.java | 8 +- .../CassandraBufferedRateReadExecutor.java | 9 +- .../CassandraBufferedRateWriteExecutor.java | 9 +- .../dao/service/RateLimitValidator.java | 4 +- .../util/AbstractBufferedRateExecutor.java | 57 +++++----- 12 files changed, 227 insertions(+), 104 deletions(-) rename common/data/src/main/java/org/thingsboard/server/common/data/limit/{LimitedApiEntry.java => RateLimitEntry.java} (79%) rename common/data/src/main/java/org/thingsboard/server/common/data/limit/{LimitedApiUtil.java => RateLimitUtil.java} (84%) create mode 100644 common/data/src/test/java/org/thingsboard/server/common/data/limit/LimitedApiTest.java rename common/data/src/test/java/org/thingsboard/server/common/data/limit/{LimitedApiUtilTest.java => RateLimitUtilTest.java} (86%) diff --git a/application/src/test/java/org/thingsboard/server/controller/TenantProfileControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/TenantProfileControllerTest.java index 0daa56728b..8f83f2c186 100644 --- a/application/src/test/java/org/thingsboard/server/controller/TenantProfileControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/TenantProfileControllerTest.java @@ -36,9 +36,11 @@ import org.thingsboard.server.common.data.queue.SubmitStrategyType; import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration; +import org.thingsboard.server.common.data.validation.RateLimit; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.queue.TbQueueCallback; +import java.lang.reflect.Field; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -314,6 +316,42 @@ public class TenantProfileControllerTest extends AbstractControllerTest { Assert.assertEquals(1, pageData.getTotalElements()); } + @Test + public void testRateLimitValidationAllFields() throws Exception { + loginSysAdmin(); + Mockito.reset(tbClusterService); + + List failedFields = new ArrayList<>(); + + for (Field field : DefaultTenantProfileConfiguration.class.getDeclaredFields()) { + RateLimit rateLimit = field.getAnnotation(RateLimit.class); + if (rateLimit == null) continue; + + String fieldName = field.getName(); + String expectedLabel = rateLimit.fieldName(); + + + TenantProfile tenantProfile = createTenantProfile("Invalid RateLimit - " + fieldName); + DefaultTenantProfileConfiguration config = (DefaultTenantProfileConfiguration) tenantProfile.getProfileData().getConfiguration(); + + field.setAccessible(true); + field.set(config, "10:1,10:1"); // Set invalid duplicate value + + try { + doPost("/api/tenantProfile", tenantProfile) + .andExpect(status().isBadRequest()) + .andExpect(statusReason(containsString(expectedLabel + " rate limit has duplicate 'Per seconds' configuration."))); + } catch (AssertionError e) { + failedFields.add(fieldName + " (label: " + expectedLabel + ")"); + } + } + + if (!failedFields.isEmpty()) { + throw new AssertionError("RateLimit validation failed for fields: " + String.join(", ", failedFields)); + } + testBroadcastEntityStateChangeEventNeverTenantProfile(); + } + private TenantProfile createTenantProfile(String name) { TenantProfile tenantProfile = new TenantProfile(); tenantProfile.setName(name); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java b/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java index 2a9991a1cc..a2582db9f3 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java @@ -21,6 +21,7 @@ import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileCon import java.util.Optional; import java.util.function.Function; +@Getter public enum LimitedApi { ENTITY_EXPORT(DefaultTenantProfileConfiguration::getTenantEntityExportRateLimit, "entity version creation", true), @@ -35,11 +36,11 @@ public enum LimitedApi { CASSANDRA_WRITE_QUERIES_RULE_ENGINE(DefaultTenantProfileConfiguration::getCassandraWriteQueryTenantRuleEngineRateLimits, "Rule Engine telemetry Cassandra write queries", true), CASSANDRA_READ_QUERIES_RULE_ENGINE(DefaultTenantProfileConfiguration::getCassandraReadQueryTenantRuleEngineRateLimits, "Rule Engine telemetry Cassandra read queries", true), CASSANDRA_READ_QUERIES_MONOLITH( - LimitedApiUtil.merge( + RateLimitUtil.merge( DefaultTenantProfileConfiguration::getCassandraReadQueryTenantCoreRateLimits, DefaultTenantProfileConfiguration::getCassandraReadQueryTenantRuleEngineRateLimits), "Telemetry read queries", true), CASSANDRA_WRITE_QUERIES_MONOLITH( - LimitedApiUtil.merge( + RateLimitUtil.merge( DefaultTenantProfileConfiguration::getCassandraWriteQueryTenantCoreRateLimits, DefaultTenantProfileConfiguration::getCassandraWriteQueryTenantRuleEngineRateLimits), "Telemetry write queries", true), EDGE_EVENTS(DefaultTenantProfileConfiguration::getEdgeEventRateLimits, "Edge events", true), @@ -58,11 +59,8 @@ public enum LimitedApi { CALCULATED_FIELD_DEBUG_EVENTS("calculated field debug events", true); private final Function configExtractor; - @Getter private final boolean perTenant; - @Getter private final boolean refillRateLimitIntervally; - @Getter private final String label; LimitedApi(Function configExtractor, String label, boolean perTenant) { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApiEntry.java b/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitEntry.java similarity index 79% rename from common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApiEntry.java rename to common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitEntry.java index 16084b7af9..566ff96c75 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApiEntry.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitEntry.java @@ -15,11 +15,11 @@ */ package org.thingsboard.server.common.data.limit; -public record LimitedApiEntry(long capacity, long durationSeconds) { +public record RateLimitEntry(long capacity, long durationSeconds) { - public static LimitedApiEntry parse(String s) { + public static RateLimitEntry parse(String s) { String[] parts = s.split(":"); - return new LimitedApiEntry(Long.parseLong(parts[0]), Long.parseLong(parts[1])); + return new RateLimitEntry(Long.parseLong(parts[0]), Long.parseLong(parts[1])); } @Override diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApiUtil.java b/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitUtil.java similarity index 84% rename from common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApiUtil.java rename to common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitUtil.java index 9c99d065d4..f2ee164dcd 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApiUtil.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitUtil.java @@ -28,14 +28,14 @@ import java.util.Set; import java.util.function.Function; import java.util.stream.Collectors; -public class LimitedApiUtil { +public class RateLimitUtil { - public static List parseConfig(String config) { + public static List parseConfig(String config) { if (config == null || config.isEmpty()) { return Collections.emptyList(); } return Arrays.stream(config.split(",")) - .map(LimitedApiEntry::parse) + .map(RateLimitEntry::parse) .toList(); } @@ -45,18 +45,18 @@ public class LimitedApiUtil { return config -> { String config1 = configExtractor1.apply(config); String config2 = configExtractor2.apply(config); - return LimitedApiUtil.mergeStrConfigs(config1, config2); // merges the configs + return RateLimitUtil.mergeStrConfigs(config1, config2); // merges the configs }; } private static String mergeStrConfigs(String firstConfig, String secondConfig) { - List all = new ArrayList<>(); + List all = new ArrayList<>(); all.addAll(parseConfig(firstConfig)); all.addAll(parseConfig(secondConfig)); Map merged = new HashMap<>(); - for (LimitedApiEntry entry : all) { + for (RateLimitEntry entry : all) { merged.merge(entry.durationSeconds(), entry.capacity(), Long::sum); } @@ -67,9 +67,9 @@ public class LimitedApiUtil { } public static boolean isValid(String configStr) { - List limitedApiEntries = parseConfig(configStr); + List limitedApiEntries = parseConfig(configStr); Set distinctDurations = new HashSet<>(); - for (LimitedApiEntry entry : limitedApiEntries) { + for (RateLimitEntry entry : limitedApiEntries) { if (!distinctDurations.add(entry.durationSeconds())) { return false; } @@ -85,7 +85,7 @@ public class LimitedApiUtil { Set distinctDurations = new HashSet<>(); return parseConfig(configStr).stream() .filter(entry -> distinctDurations.add(entry.durationSeconds())) - .map(LimitedApiEntry::toString) + .map(RateLimitEntry::toString) .collect(Collectors.joining(",")); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java index 00fecfe800..a4ff47c340 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java @@ -24,7 +24,7 @@ import lombok.NoArgsConstructor; import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.TenantProfileType; -import org.thingsboard.server.common.data.limit.LimitedApiUtil; +import org.thingsboard.server.common.data.limit.RateLimitUtil; import org.thingsboard.server.common.data.validation.RateLimit; import java.io.Serial; @@ -238,41 +238,41 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura @Deprecated(forRemoval = true, since = "4.1") public void deduplicateRateLimitsConfigs() { - this.transportTenantMsgRateLimit = LimitedApiUtil.deduplicateByDuration(transportTenantMsgRateLimit); - this.transportTenantTelemetryMsgRateLimit = LimitedApiUtil.deduplicateByDuration(transportTenantTelemetryMsgRateLimit); - this.transportTenantTelemetryDataPointsRateLimit = LimitedApiUtil.deduplicateByDuration(transportTenantTelemetryDataPointsRateLimit); + this.transportTenantMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportTenantMsgRateLimit); + this.transportTenantTelemetryMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportTenantTelemetryMsgRateLimit); + this.transportTenantTelemetryDataPointsRateLimit = RateLimitUtil.deduplicateByDuration(transportTenantTelemetryDataPointsRateLimit); - this.transportDeviceMsgRateLimit = LimitedApiUtil.deduplicateByDuration(transportDeviceMsgRateLimit); - this.transportDeviceTelemetryMsgRateLimit = LimitedApiUtil.deduplicateByDuration(transportDeviceTelemetryMsgRateLimit); - this.transportDeviceTelemetryDataPointsRateLimit = LimitedApiUtil.deduplicateByDuration(transportDeviceTelemetryDataPointsRateLimit); + this.transportDeviceMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportDeviceMsgRateLimit); + this.transportDeviceTelemetryMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportDeviceTelemetryMsgRateLimit); + this.transportDeviceTelemetryDataPointsRateLimit = RateLimitUtil.deduplicateByDuration(transportDeviceTelemetryDataPointsRateLimit); - this.transportGatewayMsgRateLimit = LimitedApiUtil.deduplicateByDuration(transportGatewayMsgRateLimit); - this.transportGatewayTelemetryMsgRateLimit = LimitedApiUtil.deduplicateByDuration(transportGatewayTelemetryMsgRateLimit); - this.transportGatewayTelemetryDataPointsRateLimit = LimitedApiUtil.deduplicateByDuration(transportGatewayTelemetryDataPointsRateLimit); + this.transportGatewayMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayMsgRateLimit); + this.transportGatewayTelemetryMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayTelemetryMsgRateLimit); + this.transportGatewayTelemetryDataPointsRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayTelemetryDataPointsRateLimit); - this.transportGatewayDeviceMsgRateLimit = LimitedApiUtil.deduplicateByDuration(transportGatewayDeviceMsgRateLimit); - this.transportGatewayDeviceTelemetryMsgRateLimit = LimitedApiUtil.deduplicateByDuration(transportGatewayDeviceTelemetryMsgRateLimit); - this.transportGatewayDeviceTelemetryDataPointsRateLimit = LimitedApiUtil.deduplicateByDuration(transportGatewayDeviceTelemetryDataPointsRateLimit); + this.transportGatewayDeviceMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayDeviceMsgRateLimit); + this.transportGatewayDeviceTelemetryMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayDeviceTelemetryMsgRateLimit); + this.transportGatewayDeviceTelemetryDataPointsRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayDeviceTelemetryDataPointsRateLimit); - this.tenantEntityExportRateLimit = LimitedApiUtil.deduplicateByDuration(tenantEntityExportRateLimit); - this.tenantEntityImportRateLimit = LimitedApiUtil.deduplicateByDuration(tenantEntityImportRateLimit); - this.tenantNotificationRequestsRateLimit = LimitedApiUtil.deduplicateByDuration(tenantNotificationRequestsRateLimit); - this.tenantNotificationRequestsPerRuleRateLimit = LimitedApiUtil.deduplicateByDuration(tenantNotificationRequestsPerRuleRateLimit); + this.tenantEntityExportRateLimit = RateLimitUtil.deduplicateByDuration(tenantEntityExportRateLimit); + this.tenantEntityImportRateLimit = RateLimitUtil.deduplicateByDuration(tenantEntityImportRateLimit); + this.tenantNotificationRequestsRateLimit = RateLimitUtil.deduplicateByDuration(tenantNotificationRequestsRateLimit); + this.tenantNotificationRequestsPerRuleRateLimit = RateLimitUtil.deduplicateByDuration(tenantNotificationRequestsPerRuleRateLimit); - this.cassandraReadQueryTenantCoreRateLimits = LimitedApiUtil.deduplicateByDuration(cassandraReadQueryTenantCoreRateLimits); - this.cassandraWriteQueryTenantCoreRateLimits = LimitedApiUtil.deduplicateByDuration(cassandraWriteQueryTenantCoreRateLimits); - this.cassandraReadQueryTenantRuleEngineRateLimits = LimitedApiUtil.deduplicateByDuration(cassandraReadQueryTenantRuleEngineRateLimits); - this.cassandraWriteQueryTenantRuleEngineRateLimits = LimitedApiUtil.deduplicateByDuration(cassandraWriteQueryTenantRuleEngineRateLimits); + this.cassandraReadQueryTenantCoreRateLimits = RateLimitUtil.deduplicateByDuration(cassandraReadQueryTenantCoreRateLimits); + this.cassandraWriteQueryTenantCoreRateLimits = RateLimitUtil.deduplicateByDuration(cassandraWriteQueryTenantCoreRateLimits); + this.cassandraReadQueryTenantRuleEngineRateLimits = RateLimitUtil.deduplicateByDuration(cassandraReadQueryTenantRuleEngineRateLimits); + this.cassandraWriteQueryTenantRuleEngineRateLimits = RateLimitUtil.deduplicateByDuration(cassandraWriteQueryTenantRuleEngineRateLimits); - this.edgeEventRateLimits = LimitedApiUtil.deduplicateByDuration(edgeEventRateLimits); - this.edgeEventRateLimitsPerEdge = LimitedApiUtil.deduplicateByDuration(edgeEventRateLimitsPerEdge); - this.edgeUplinkMessagesRateLimits = LimitedApiUtil.deduplicateByDuration(edgeUplinkMessagesRateLimits); - this.edgeUplinkMessagesRateLimitsPerEdge = LimitedApiUtil.deduplicateByDuration(edgeUplinkMessagesRateLimitsPerEdge); + this.edgeEventRateLimits = RateLimitUtil.deduplicateByDuration(edgeEventRateLimits); + this.edgeEventRateLimitsPerEdge = RateLimitUtil.deduplicateByDuration(edgeEventRateLimitsPerEdge); + this.edgeUplinkMessagesRateLimits = RateLimitUtil.deduplicateByDuration(edgeUplinkMessagesRateLimits); + this.edgeUplinkMessagesRateLimitsPerEdge = RateLimitUtil.deduplicateByDuration(edgeUplinkMessagesRateLimitsPerEdge); - this.wsUpdatesPerSessionRateLimit = LimitedApiUtil.deduplicateByDuration(wsUpdatesPerSessionRateLimit); + this.wsUpdatesPerSessionRateLimit = RateLimitUtil.deduplicateByDuration(wsUpdatesPerSessionRateLimit); - this.tenantServerRestLimitsConfiguration = LimitedApiUtil.deduplicateByDuration(tenantServerRestLimitsConfiguration); - this.customerServerRestLimitsConfiguration = LimitedApiUtil.deduplicateByDuration(customerServerRestLimitsConfiguration); + this.tenantServerRestLimitsConfiguration = RateLimitUtil.deduplicateByDuration(tenantServerRestLimitsConfiguration); + this.customerServerRestLimitsConfiguration = RateLimitUtil.deduplicateByDuration(customerServerRestLimitsConfiguration); } } diff --git a/common/data/src/test/java/org/thingsboard/server/common/data/limit/LimitedApiTest.java b/common/data/src/test/java/org/thingsboard/server/common/data/limit/LimitedApiTest.java new file mode 100644 index 0000000000..d650017fd4 --- /dev/null +++ b/common/data/src/test/java/org/thingsboard/server/common/data/limit/LimitedApiTest.java @@ -0,0 +1,102 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed 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 org.thingsboard.server.common.data.limit; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; + +import java.util.Arrays; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +class LimitedApiTest { + + private DefaultTenantProfileConfiguration config; + + @BeforeEach + void setUp() { + config = mock(DefaultTenantProfileConfiguration.class); + } + + @Test + void testCorrectConfigExtractorsUsed() { + Map verifierMap = Map.ofEntries( + Map.entry(LimitedApi.ENTITY_EXPORT, () -> + verify(config).getTenantEntityExportRateLimit()), + Map.entry(LimitedApi.ENTITY_IMPORT, () -> + verify(config).getTenantEntityImportRateLimit()), + Map.entry(LimitedApi.NOTIFICATION_REQUESTS, () -> + verify(config).getTenantNotificationRequestsRateLimit()), + Map.entry(LimitedApi.NOTIFICATION_REQUESTS_PER_RULE, () -> + verify(config).getTenantNotificationRequestsPerRuleRateLimit()), + Map.entry(LimitedApi.REST_REQUESTS_PER_TENANT, () -> + verify(config).getTenantServerRestLimitsConfiguration()), + Map.entry(LimitedApi.REST_REQUESTS_PER_CUSTOMER, () -> + verify(config).getCustomerServerRestLimitsConfiguration()), + Map.entry(LimitedApi.WS_UPDATES_PER_SESSION, () -> + verify(config).getWsUpdatesPerSessionRateLimit()), + Map.entry(LimitedApi.CASSANDRA_WRITE_QUERIES_CORE, () -> + verify(config).getCassandraWriteQueryTenantCoreRateLimits()), + Map.entry(LimitedApi.CASSANDRA_READ_QUERIES_CORE, () -> + verify(config).getCassandraReadQueryTenantCoreRateLimits()), + Map.entry(LimitedApi.CASSANDRA_WRITE_QUERIES_RULE_ENGINE, () -> + verify(config).getCassandraWriteQueryTenantRuleEngineRateLimits()), + Map.entry(LimitedApi.CASSANDRA_READ_QUERIES_RULE_ENGINE, () -> + verify(config).getCassandraReadQueryTenantRuleEngineRateLimits()), + Map.entry(LimitedApi.CASSANDRA_READ_QUERIES_MONOLITH, () -> { + verify(config).getCassandraReadQueryTenantCoreRateLimits(); + verify(config).getCassandraReadQueryTenantRuleEngineRateLimits(); + }), + Map.entry(LimitedApi.CASSANDRA_WRITE_QUERIES_MONOLITH, () -> { + verify(config).getCassandraWriteQueryTenantCoreRateLimits(); + verify(config).getCassandraWriteQueryTenantRuleEngineRateLimits(); + }), + Map.entry(LimitedApi.EDGE_EVENTS, () -> + verify(config).getEdgeEventRateLimits()), + Map.entry(LimitedApi.EDGE_EVENTS_PER_EDGE, () -> + verify(config).getEdgeEventRateLimitsPerEdge()), + Map.entry(LimitedApi.EDGE_UPLINK_MESSAGES, () -> + verify(config).getEdgeUplinkMessagesRateLimits()), + Map.entry(LimitedApi.EDGE_UPLINK_MESSAGES_PER_EDGE, () -> + verify(config).getEdgeUplinkMessagesRateLimitsPerEdge()) + ); + + Set expected = verifierMap.keySet(); + Set actual = Arrays.stream(LimitedApi.values()) + .filter(api -> api.getConfigExtractor() != null) + .collect(Collectors.toSet()); + + assertThat(expected) + .as("Verifier map should cover all LimitedApis with extractors") + .containsExactlyInAnyOrderElementsOf(actual); + + for (Map.Entry entry : verifierMap.entrySet()) { + LimitedApi api = entry.getKey(); + api.getLimitConfig(config); + entry.getValue().run(); + clearInvocations(config); + } + } + + +} \ No newline at end of file diff --git a/common/data/src/test/java/org/thingsboard/server/common/data/limit/LimitedApiUtilTest.java b/common/data/src/test/java/org/thingsboard/server/common/data/limit/RateLimitUtilTest.java similarity index 86% rename from common/data/src/test/java/org/thingsboard/server/common/data/limit/LimitedApiUtilTest.java rename to common/data/src/test/java/org/thingsboard/server/common/data/limit/RateLimitUtilTest.java index c275df790c..a410785126 100644 --- a/common/data/src/test/java/org/thingsboard/server/common/data/limit/LimitedApiUtilTest.java +++ b/common/data/src/test/java/org/thingsboard/server/common/data/limit/RateLimitUtilTest.java @@ -24,12 +24,12 @@ import java.util.function.Function; import static org.assertj.core.api.Assertions.assertThat; -class LimitedApiUtilTest { +class RateLimitUtilTest { @Test @DisplayName("LimitedApiUtil should parse single entry correctly") void testParseSingleEntry() { - List entries = LimitedApiUtil.parseConfig("100:60"); + List entries = RateLimitUtil.parseConfig("100:60"); assertThat(entries).hasSize(1); assertThat(entries.get(0).capacity()).isEqualTo(100); @@ -39,7 +39,7 @@ class LimitedApiUtilTest { @Test @DisplayName("LimitedApiUtil should parse multiple entries correctly") void testParseMultipleEntries() { - List entries = LimitedApiUtil.parseConfig("100:60,200:30"); + List entries = RateLimitUtil.parseConfig("100:60,200:30"); assertThat(entries).hasSize(2); assertThat(entries.get(0).capacity()).isEqualTo(100); @@ -51,8 +51,8 @@ class LimitedApiUtilTest { @Test @DisplayName("LimitedApiUtil should return empty list for null or empty config") void testParseEmptyConfig() { - assertThat(LimitedApiUtil.parseConfig(null)).isEmpty(); - assertThat(LimitedApiUtil.parseConfig("")).isEmpty(); + assertThat(RateLimitUtil.parseConfig(null)).isEmpty(); + assertThat(RateLimitUtil.parseConfig("")).isEmpty(); } @Test @@ -64,7 +64,7 @@ class LimitedApiUtilTest { // Fake config instance (not used directly in lambda logic) DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); - String result = LimitedApiUtil.merge(extractor1, extractor2).apply(config); + String result = RateLimitUtil.merge(extractor1, extractor2).apply(config); // Should be: 300:60 (100+200), 50:30, 25:10 assertThat(result).isEqualTo("25:10,50:30,300:60"); @@ -78,7 +78,7 @@ class LimitedApiUtilTest { // Fake config instance (not used directly in lambda logic) DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); - String result = LimitedApiUtil.merge(extractor1, extractor2).apply(config); + String result = RateLimitUtil.merge(extractor1, extractor2).apply(config); assertThat(result).isEqualTo("100:60"); } @@ -91,7 +91,7 @@ class LimitedApiUtilTest { // Fake config instance (not used directly in lambda logic) DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); - String result = LimitedApiUtil.merge(extractor1, extractor2).apply(config); + String result = RateLimitUtil.merge(extractor1, extractor2).apply(config); assertThat(result).isEqualTo("200:10,100:60"); } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/tools/TbRateLimits.java b/common/message/src/main/java/org/thingsboard/server/common/msg/tools/TbRateLimits.java index c65c90706a..182cb21ab8 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/tools/TbRateLimits.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/tools/TbRateLimits.java @@ -21,8 +21,8 @@ import io.github.bucket4j.Bucket; import io.github.bucket4j.local.LocalBucket; import io.github.bucket4j.local.LocalBucketBuilder; import lombok.Getter; -import org.thingsboard.server.common.data.limit.LimitedApiEntry; -import org.thingsboard.server.common.data.limit.LimitedApiUtil; +import org.thingsboard.server.common.data.limit.RateLimitEntry; +import org.thingsboard.server.common.data.limit.RateLimitUtil; import java.time.Duration; import java.util.List; @@ -41,12 +41,12 @@ public class TbRateLimits { } public TbRateLimits(String limitsConfiguration, boolean refillIntervally) { - List limitedApiEntries = LimitedApiUtil.parseConfig(limitsConfiguration); + List limitedApiEntries = RateLimitUtil.parseConfig(limitsConfiguration); if (limitedApiEntries.isEmpty()) { throw new IllegalArgumentException("Failed to parse rate limits configuration: " + limitsConfiguration); } LocalBucketBuilder localBucket = Bucket.builder(); - for (LimitedApiEntry entry : limitedApiEntries) { + for (RateLimitEntry entry : limitedApiEntries) { BandwidthBuilder.BandwidthBuilderRefillStage bandwidthBuilder = Bandwidth.builder().capacity(entry.capacity()); Bandwidth bandwidth = refillIntervally ? bandwidthBuilder.refillIntervally(entry.capacity(), Duration.ofSeconds(entry.durationSeconds())).build() : diff --git a/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraBufferedRateReadExecutor.java b/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraBufferedRateReadExecutor.java index 98cfef7f7d..8698429b60 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraBufferedRateReadExecutor.java +++ b/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraBufferedRateReadExecutor.java @@ -53,8 +53,8 @@ public class CassandraBufferedRateReadExecutor extends AbstractBufferedRateExecu @Autowired EntityService entityService, @Autowired RateLimitService rateLimitService, @Autowired(required = false) TbServiceInfoProvider serviceInfoProvider) { - super(queueLimit, concurrencyLimit, maxWaitTime, dispatcherThreads, callbackThreads, pollMs, printQueriesFreq, statsFactory, - entityService, rateLimitService, serviceInfoProvider, printTenantNames); + super(queueLimit, concurrencyLimit, maxWaitTime, dispatcherThreads, callbackThreads, pollMs, printQueriesFreq, + BufferedRateExecutorType.READ, entityService, rateLimitService, serviceInfoProvider, statsFactory, printTenantNames); } @Scheduled(fixedDelayString = "${cassandra.query.rate_limit_print_interval_ms}") @@ -68,11 +68,6 @@ public class CassandraBufferedRateReadExecutor extends AbstractBufferedRateExecu super.stop(); } - @Override - protected BufferedRateExecutorType getBufferedRateExecutorType() { - return BufferedRateExecutorType.READ; - } - @Override protected SettableFuture create() { return SettableFuture.create(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraBufferedRateWriteExecutor.java b/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraBufferedRateWriteExecutor.java index 1fb4b8dea5..6eb313be96 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraBufferedRateWriteExecutor.java +++ b/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraBufferedRateWriteExecutor.java @@ -53,8 +53,8 @@ public class CassandraBufferedRateWriteExecutor extends AbstractBufferedRateExec @Autowired EntityService entityService, @Autowired RateLimitService rateLimitService, @Autowired(required = false) TbServiceInfoProvider serviceInfoProvider) { - super(queueLimit, concurrencyLimit, maxWaitTime, dispatcherThreads, callbackThreads, pollMs, printQueriesFreq, statsFactory, - entityService, rateLimitService, serviceInfoProvider, printTenantNames); + super(queueLimit, concurrencyLimit, maxWaitTime, dispatcherThreads, callbackThreads, pollMs, printQueriesFreq, + BufferedRateExecutorType.WRITE, entityService, rateLimitService, serviceInfoProvider, statsFactory, printTenantNames); } @Scheduled(fixedDelayString = "${cassandra.query.rate_limit_print_interval_ms}") @@ -68,11 +68,6 @@ public class CassandraBufferedRateWriteExecutor extends AbstractBufferedRateExec super.stop(); } - @Override - protected BufferedRateExecutorType getBufferedRateExecutorType() { - return BufferedRateExecutorType.WRITE; - } - @Override protected SettableFuture create() { return SettableFuture.create(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/RateLimitValidator.java b/dao/src/main/java/org/thingsboard/server/dao/service/RateLimitValidator.java index 0703f8b65d..bf776d4dc7 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/service/RateLimitValidator.java +++ b/dao/src/main/java/org/thingsboard/server/dao/service/RateLimitValidator.java @@ -18,7 +18,7 @@ package org.thingsboard.server.dao.service; import jakarta.validation.ConstraintValidator; import jakarta.validation.ConstraintValidatorContext; import lombok.extern.slf4j.Slf4j; -import org.thingsboard.server.common.data.limit.LimitedApiUtil; +import org.thingsboard.server.common.data.limit.RateLimitUtil; import org.thingsboard.server.common.data.validation.RateLimit; @Slf4j @@ -26,7 +26,7 @@ public class RateLimitValidator implements ConstraintValidator> queue; private final ExecutorService dispatcherExecutor; private final ExecutorService callbackExecutor; @@ -80,29 +81,32 @@ public abstract class AbstractBufferedRateExecutor tenantNamesCache = new HashMap<>(); + private final LimitedApi myLimitedApi; + public AbstractBufferedRateExecutor(int queueLimit, int concurrencyLimit, long maxWaitTime, int dispatcherThreads, - int callbackThreads, long pollMs, int printQueriesFreq, StatsFactory statsFactory, - EntityService entityService, RateLimitService rateLimitService, TbServiceInfoProvider serviceInfoProvider, boolean printTenantNames) { + int callbackThreads, long pollMs, int printQueriesFreq, BufferedRateExecutorType executorType, + EntityService entityService, RateLimitService rateLimitService, TbServiceInfoProvider serviceInfoProvider, + StatsFactory statsFactory, boolean printTenantNames) { this.maxWaitTime = maxWaitTime; this.pollMs = pollMs; + this.bufferName = executorType.getDisplayName(); this.concurrencyLimit = concurrencyLimit; this.printQueriesFreq = printQueriesFreq; this.queue = new LinkedBlockingDeque<>(queueLimit); - this.dispatcherExecutor = Executors.newFixedThreadPool(dispatcherThreads, ThingsBoardThreadFactory.forName("nosql-" + getBufferName() + "-dispatcher")); - this.callbackExecutor = ThingsBoardExecutors.newWorkStealingPool(callbackThreads, "nosql-" + getBufferName() + "-callback"); - this.timeoutExecutor = ThingsBoardExecutors.newSingleThreadScheduledExecutor("nosql-" + getBufferName() + "-timeout"); + this.dispatcherExecutor = Executors.newFixedThreadPool(dispatcherThreads, ThingsBoardThreadFactory.forName("nosql-" + bufferName + "-dispatcher")); + this.callbackExecutor = ThingsBoardExecutors.newWorkStealingPool(callbackThreads, "nosql-" + bufferName + "-callback"); + this.timeoutExecutor = ThingsBoardExecutors.newSingleThreadScheduledExecutor("nosql-" + bufferName + "-timeout"); this.stats = new BufferedRateExecutorStats(statsFactory); - String concurrencyLevelKey = StatsType.RATE_EXECUTOR.getName() + "." + CONCURRENCY_LEVEL + getBufferName(); //metric name may change with buffer name suffix + String concurrencyLevelKey = StatsType.RATE_EXECUTOR.getName() + "." + CONCURRENCY_LEVEL + bufferName; //metric name may change with buffer name suffix this.concurrencyLevel = statsFactory.createGauge(concurrencyLevelKey, new AtomicInteger(0)); this.entityService = entityService; this.rateLimitService = rateLimitService; - this.serviceInfoProvider = serviceInfoProvider; + this.myLimitedApi = resolveLimitedApi(serviceInfoProvider, executorType); this.printTenantNames = printTenantNames; for (int i = 0; i < dispatcherThreads; i++) { @@ -118,14 +122,14 @@ public abstract class AbstractBufferedRateExecutor execute(AsyncTaskContext taskCtx); - private String getBufferName() { - return getBufferedRateExecutorType().getDisplayName(); - } - - protected abstract BufferedRateExecutorType getBufferedRateExecutorType(); - private void dispatch() { - log.info("[{}] Buffered rate executor thread started", getBufferName()); + log.info("[{}] Buffered rate executor thread started", bufferName); while (!Thread.interrupted()) { int curLvl = concurrencyLevel.get(); AsyncTaskContext taskCtx = null; @@ -190,7 +185,7 @@ public abstract class AbstractBufferedRateExecutor= printQueriesFreq) { printQueriesIdx.set(0); String query = queryToString(finalTaskCtx); - log.info("[{}][{}] Cassandra query: {}", getBufferName(), taskCtx.getId(), query); + log.info("[{}][{}] Cassandra query: {}", bufferName, taskCtx.getId(), query); } } logTask("Processing", finalTaskCtx); @@ -243,7 +238,7 @@ public abstract class AbstractBufferedRateExecutor taskCtx) { @@ -319,7 +314,7 @@ public abstract class AbstractBufferedRateExecutor