diff --git a/application/pom.xml b/application/pom.xml index 557179e918..34218956fc 100644 --- a/application/pom.xml +++ b/application/pom.xml @@ -128,6 +128,10 @@ org.thingsboard.common edqs + + org.thingsboard.common + discovery-api + org.thingsboard dao diff --git a/application/src/main/data/upgrade/basic/schema_update.sql b/application/src/main/data/upgrade/basic/schema_update.sql index 016e786776..ecb2a147bc 100644 --- a/application/src/main/data/upgrade/basic/schema_update.sql +++ b/application/src/main/data/upgrade/basic/schema_update.sql @@ -14,3 +14,31 @@ -- limitations under the License. -- +UPDATE tenant_profile +SET profile_data = jsonb_set( + profile_data, + '{configuration}', + ( + (profile_data -> 'configuration') - 'cassandraQueryTenantRateLimitsConfiguration' + || + COALESCE( + CASE + WHEN profile_data -> 'configuration' -> + 'cassandraQueryTenantRateLimitsConfiguration' IS NOT NULL THEN + jsonb_build_object( + 'cassandraReadQueryTenantCoreRateLimits', + profile_data -> 'configuration' -> 'cassandraQueryTenantRateLimitsConfiguration', + 'cassandraWriteQueryTenantCoreRateLimits', + profile_data -> 'configuration' -> 'cassandraQueryTenantRateLimitsConfiguration', + 'cassandraReadQueryTenantRuleEngineRateLimits', + profile_data -> 'configuration' -> 'cassandraQueryTenantRateLimitsConfiguration', + 'cassandraWriteQueryTenantRuleEngineRateLimits', + profile_data -> 'configuration' -> 'cassandraQueryTenantRateLimitsConfiguration' + ) + END, + '{}'::jsonb + ) + ) + ) +WHERE profile_data -> 'configuration' ? 'cassandraQueryTenantRateLimitsConfiguration'; + diff --git a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java index d8ddd7cd9f..c3f1cee046 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java @@ -24,6 +24,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Profile; import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.alarm.AlarmSeverity; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleNodeId; @@ -34,8 +35,10 @@ import org.thingsboard.server.common.data.query.FilterPredicateValue; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.common.data.rule.RuleNode; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rule.RuleChainService; +import org.thingsboard.server.dao.tenant.TenantProfileService; import org.thingsboard.server.service.component.ComponentDiscoveryService; import org.thingsboard.server.service.component.RuleNodeClassInfo; import org.thingsboard.server.service.install.DbUpgradeExecutorService; @@ -69,14 +72,57 @@ public class DefaultDataUpdateService implements DataUpdateService { @Autowired private DbUpgradeExecutorService executorService; + @Autowired + private TenantProfileService tenantProfileService; + @Override public void updateData() throws Exception { log.info("Updating data ..."); //TODO: should be cleaned after each release updateInputNodes(); + deduplicateRateLimitsPerSecondsConfigurations(); log.info("Data updated."); } + private void deduplicateRateLimitsPerSecondsConfigurations() { + log.info("Starting update of tenant profiles..."); + + int totalProfiles = 0; + int updatedTenantProfiles = 0; + int skippedProfiles = 0; + int failedProfiles = 0; + + var tenantProfiles = new PageDataIterable<>( + pageLink -> tenantProfileService.findTenantProfiles(TenantId.SYS_TENANT_ID, pageLink), 1024); + + for (TenantProfile tenantProfile : tenantProfiles) { + totalProfiles++; + String profileName = tenantProfile.getName(); + UUID profileId = tenantProfile.getId().getId(); + try { + Optional profileConfiguration = tenantProfile.getProfileConfiguration(); + if (profileConfiguration.isEmpty()) { + log.debug("[{}][{}] Skipping tenant profile with non-default configuration.", profileId, profileName); + skippedProfiles++; + continue; + } + + DefaultTenantProfileConfiguration defaultTenantProfileConfiguration = profileConfiguration.get(); + defaultTenantProfileConfiguration.deduplicateRateLimitsConfigs(); + tenantProfileService.saveTenantProfile(TenantId.SYS_TENANT_ID, tenantProfile); + updatedTenantProfiles++; + log.debug("[{}][{}] Successfully updated tenant profile.", profileId, profileName); + } catch (Exception e) { + log.error("[{}][{}] Failed to updated tenant profile: ", profileId, profileName, e); + failedProfiles++; + } + } + + log.info("Tenant profiles update completed. Total: {}, Updated: {}, Skipped: {}, Failed: {}", + totalProfiles, updatedTenantProfiles, skippedProfiles, failedProfiles); + } + + private void updateInputNodes() { log.info("Creating relations for input nodes..."); int n = 0; 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/application/src/test/java/org/thingsboard/server/service/limits/RateLimitServiceTest.java b/application/src/test/java/org/thingsboard/server/service/limits/RateLimitServiceTest.java index 3eb5793c74..674b7bfdb9 100644 --- a/application/src/test/java/org/thingsboard/server/service/limits/RateLimitServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/limits/RateLimitServiceTest.java @@ -52,7 +52,7 @@ public class RateLimitServiceTest { public void beforeEach() { tenantProfileCache = Mockito.mock(DefaultTbTenantProfileCache.class); rateLimitService = new DefaultRateLimitService(tenantProfileCache, mock(NotificationRuleProcessor.class), 60, 100); - tenantId = new TenantId(UUID.randomUUID()); + tenantId = TenantId.fromUUID(UUID.randomUUID()); } @Test @@ -67,7 +67,10 @@ public class RateLimitServiceTest { profileConfiguration.setTenantServerRestLimitsConfiguration(rateLimit); profileConfiguration.setCustomerServerRestLimitsConfiguration(rateLimit); profileConfiguration.setWsUpdatesPerSessionRateLimit(rateLimit); - profileConfiguration.setCassandraQueryTenantRateLimitsConfiguration(rateLimit); + profileConfiguration.setCassandraReadQueryTenantCoreRateLimits(rateLimit); + profileConfiguration.setCassandraWriteQueryTenantCoreRateLimits(rateLimit); + profileConfiguration.setCassandraReadQueryTenantRuleEngineRateLimits(rateLimit); + profileConfiguration.setCassandraWriteQueryTenantRuleEngineRateLimits(rateLimit); profileConfiguration.setEdgeEventRateLimits(rateLimit); profileConfiguration.setEdgeEventRateLimitsPerEdge(rateLimit); profileConfiguration.setEdgeUplinkMessagesRateLimits(rateLimit); @@ -79,7 +82,10 @@ public class RateLimitServiceTest { LimitedApi.ENTITY_IMPORT, LimitedApi.NOTIFICATION_REQUESTS, LimitedApi.REST_REQUESTS_PER_CUSTOMER, - LimitedApi.CASSANDRA_QUERIES, + LimitedApi.CASSANDRA_READ_QUERIES_CORE, + LimitedApi.CASSANDRA_WRITE_QUERIES_CORE, + LimitedApi.CASSANDRA_READ_QUERIES_RULE_ENGINE, + LimitedApi.CASSANDRA_WRITE_QUERIES_RULE_ENGINE, LimitedApi.EDGE_EVENTS, LimitedApi.EDGE_EVENTS_PER_EDGE, LimitedApi.EDGE_UPLINK_MESSAGES, @@ -88,6 +94,13 @@ public class RateLimitServiceTest { testRateLimits(limitedApi, max, tenantId); } + for (LimitedApi limitedApi : List.of( + LimitedApi.CASSANDRA_READ_QUERIES_MONOLITH, + LimitedApi.CASSANDRA_WRITE_QUERIES_MONOLITH + )) { + testRateLimits(limitedApi, max * 2, tenantId); + } + CustomerId customerId = new CustomerId(UUID.randomUUID()); testRateLimits(LimitedApi.REST_REQUESTS_PER_CUSTOMER, max, customerId); diff --git a/common/coap-server/pom.xml b/common/coap-server/pom.xml index fe6ae3c64c..b5ca9c1b9c 100644 --- a/common/coap-server/pom.xml +++ b/common/coap-server/pom.xml @@ -46,6 +46,10 @@ org.thingsboard.common data + + org.thingsboard.common + discovery-api + org.thingsboard.common.transport transport-api 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 db7f14171b..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), @@ -30,7 +31,18 @@ public enum LimitedApi { REST_REQUESTS_PER_TENANT(DefaultTenantProfileConfiguration::getTenantServerRestLimitsConfiguration, "REST API requests", true), REST_REQUESTS_PER_CUSTOMER(DefaultTenantProfileConfiguration::getCustomerServerRestLimitsConfiguration, "REST API requests per customer", false), WS_UPDATES_PER_SESSION(DefaultTenantProfileConfiguration::getWsUpdatesPerSessionRateLimit, "WS updates per session", true), - CASSANDRA_QUERIES(DefaultTenantProfileConfiguration::getCassandraQueryTenantRateLimitsConfiguration, "Cassandra queries", true), + CASSANDRA_WRITE_QUERIES_CORE(DefaultTenantProfileConfiguration::getCassandraWriteQueryTenantCoreRateLimits, "Rest API Cassandra write queries", true), + CASSANDRA_READ_QUERIES_CORE(DefaultTenantProfileConfiguration::getCassandraReadQueryTenantCoreRateLimits, "Rest API and WS telemetry Cassandra read queries", true), + 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( + RateLimitUtil.merge( + DefaultTenantProfileConfiguration::getCassandraReadQueryTenantCoreRateLimits, + DefaultTenantProfileConfiguration::getCassandraReadQueryTenantRuleEngineRateLimits), "Telemetry read queries", true), + CASSANDRA_WRITE_QUERIES_MONOLITH( + RateLimitUtil.merge( + DefaultTenantProfileConfiguration::getCassandraWriteQueryTenantCoreRateLimits, + DefaultTenantProfileConfiguration::getCassandraWriteQueryTenantRuleEngineRateLimits), "Telemetry write queries", true), EDGE_EVENTS(DefaultTenantProfileConfiguration::getEdgeEventRateLimits, "Edge events", true), EDGE_EVENTS_PER_EDGE(DefaultTenantProfileConfiguration::getEdgeEventRateLimitsPerEdge, "Edge events per edge", false), EDGE_UPLINK_MESSAGES(DefaultTenantProfileConfiguration::getEdgeUplinkMessagesRateLimits, "Edge uplink messages", true), @@ -46,28 +58,28 @@ public enum LimitedApi { WS_SUBSCRIPTIONS("WS subscriptions", false), CALCULATED_FIELD_DEBUG_EVENTS("calculated field debug events", true); - private Function configExtractor; - @Getter + private final Function configExtractor; private final boolean perTenant; - @Getter - private boolean refillRateLimitIntervally; - @Getter - private String label; + private final boolean refillRateLimitIntervally; + private final String label; LimitedApi(Function configExtractor, String label, boolean perTenant) { - this.configExtractor = configExtractor; - this.label = label; - this.perTenant = perTenant; + this(configExtractor, label, perTenant, false); } LimitedApi(boolean perTenant, boolean refillRateLimitIntervally) { - this.perTenant = perTenant; - this.refillRateLimitIntervally = refillRateLimitIntervally; + this(null, null, perTenant, refillRateLimitIntervally); } LimitedApi(String label, boolean perTenant) { + this(null, label, perTenant, false); + } + + LimitedApi(Function configExtractor, String label, boolean perTenant, boolean refillRateLimitIntervally) { + this.configExtractor = configExtractor; this.label = label; this.perTenant = perTenant; + this.refillRateLimitIntervally = refillRateLimitIntervally; } public String getLimitConfig(DefaultTenantProfileConfiguration profileConfiguration) { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitEntry.java b/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitEntry.java new file mode 100644 index 0000000000..566ff96c75 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitEntry.java @@ -0,0 +1,30 @@ +/** + * 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; + +public record RateLimitEntry(long capacity, long durationSeconds) { + + public static RateLimitEntry parse(String s) { + String[] parts = s.split(":"); + return new RateLimitEntry(Long.parseLong(parts[0]), Long.parseLong(parts[1])); + } + + @Override + public String toString() { + return capacity + ":" + durationSeconds; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitUtil.java b/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitUtil.java new file mode 100644 index 0000000000..f2ee164dcd --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/limit/RateLimitUtil.java @@ -0,0 +1,92 @@ +/** + * 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.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; + +public class RateLimitUtil { + + public static List parseConfig(String config) { + if (config == null || config.isEmpty()) { + return Collections.emptyList(); + } + return Arrays.stream(config.split(",")) + .map(RateLimitEntry::parse) + .toList(); + } + + public static Function merge( + Function configExtractor1, + Function configExtractor2) { + return config -> { + String config1 = configExtractor1.apply(config); + String config2 = configExtractor2.apply(config); + return RateLimitUtil.mergeStrConfigs(config1, config2); // merges the configs + }; + } + + private static String mergeStrConfigs(String firstConfig, String secondConfig) { + List all = new ArrayList<>(); + all.addAll(parseConfig(firstConfig)); + all.addAll(parseConfig(secondConfig)); + + Map merged = new HashMap<>(); + + for (RateLimitEntry entry : all) { + merged.merge(entry.durationSeconds(), entry.capacity(), Long::sum); + } + + return merged.entrySet().stream() + .sorted(Map.Entry.comparingByKey()) // optional: sort by duration + .map(e -> e.getValue() + ":" + e.getKey()) + .collect(Collectors.joining(",")); + } + + public static boolean isValid(String configStr) { + List limitedApiEntries = parseConfig(configStr); + Set distinctDurations = new HashSet<>(); + for (RateLimitEntry entry : limitedApiEntries) { + if (!distinctDurations.add(entry.durationSeconds())) { + return false; + } + } + return true; + } + + @Deprecated(forRemoval = true, since = "4.1") + public static String deduplicateByDuration(String configStr) { + if (configStr == null) { + return null; + } + Set distinctDurations = new HashSet<>(); + return parseConfig(configStr).stream() + .filter(entry -> distinctDurations.add(entry.durationSeconds())) + .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 f256b02d9a..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,6 +24,8 @@ 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.RateLimitUtil; +import org.thingsboard.server.common.data.validation.RateLimit; import java.io.Serial; @@ -49,37 +51,53 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura private long maxResourceSize; @Schema(example = "1000:1,20000:60") + @RateLimit(fieldName = "Transport tenant messages") private String transportTenantMsgRateLimit; @Schema(example = "1000:1,20000:60") + @RateLimit(fieldName = "Transport tenant telemetry messages") private String transportTenantTelemetryMsgRateLimit; @Schema(example = "1000:1,20000:60") + @RateLimit(fieldName = "Transport tenant telemetry data points") private String transportTenantTelemetryDataPointsRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Transport device messages") private String transportDeviceMsgRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Transport device telemetry messages") private String transportDeviceTelemetryMsgRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Transport device telemetry data points") private String transportDeviceTelemetryDataPointsRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Transport gateway messages") private String transportGatewayMsgRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Transport gateway telemetry messages") private String transportGatewayTelemetryMsgRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Transport gateway telemetry data points") private String transportGatewayTelemetryDataPointsRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Transport gateway device messages") private String transportGatewayDeviceMsgRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Transport gateway device telemetry messages") private String transportGatewayDeviceTelemetryMsgRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Transport gateway device telemetry data points") private String transportGatewayDeviceTelemetryDataPointsRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Entity version creation") private String tenantEntityExportRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Entity version load") private String tenantEntityImportRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Notification requests") private String tenantNotificationRequestsRateLimit; @Schema(example = "20:1,600:60") + @RateLimit(fieldName = "Notification requests per notification rule") private String tenantNotificationRequestsPerRuleRateLimit; @Schema(example = "10000000") @@ -107,7 +125,9 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura @Schema(example = "1000") private long maxCreatedAlarms; + @RateLimit(fieldName = "REST requests for tenant") private String tenantServerRestLimitsConfiguration; + @RateLimit(fieldName = "REST requests for customer") private String customerServerRestLimitsConfiguration; private int maxWsSessionsPerTenant; @@ -119,13 +139,26 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura private long maxWsSubscriptionsPerCustomer; private long maxWsSubscriptionsPerRegularUser; private long maxWsSubscriptionsPerPublicUser; + @RateLimit(fieldName = "WS updates per session") private String wsUpdatesPerSessionRateLimit; - private String cassandraQueryTenantRateLimitsConfiguration; + @RateLimit(fieldName = "Rest API and WS telemetry Cassandra read queries") + private String cassandraReadQueryTenantCoreRateLimits; + @RateLimit(fieldName = "Rest API Cassandra write queries") + private String cassandraWriteQueryTenantCoreRateLimits; + @RateLimit(fieldName = "Rule Engine telemetry Cassandra read queries") + private String cassandraReadQueryTenantRuleEngineRateLimits; + @RateLimit(fieldName = "Rule Engine telemetry Cassandra write queries") + private String cassandraWriteQueryTenantRuleEngineRateLimits; + + @RateLimit(fieldName = "Edge events") private String edgeEventRateLimits; + @RateLimit(fieldName = "Edge events per edge") private String edgeEventRateLimitsPerEdge; + @RateLimit(fieldName = "Edge uplink messages") private String edgeUplinkMessagesRateLimits; + @RateLimit(fieldName = "Edge uplink messages per edge") private String edgeUplinkMessagesRateLimitsPerEdge; private int defaultStorageTtlDays; @@ -203,4 +236,43 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura return maxRuleNodeExecutionsPerMessage; } + @Deprecated(forRemoval = true, since = "4.1") + public void deduplicateRateLimitsConfigs() { + this.transportTenantMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportTenantMsgRateLimit); + this.transportTenantTelemetryMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportTenantTelemetryMsgRateLimit); + this.transportTenantTelemetryDataPointsRateLimit = RateLimitUtil.deduplicateByDuration(transportTenantTelemetryDataPointsRateLimit); + + this.transportDeviceMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportDeviceMsgRateLimit); + this.transportDeviceTelemetryMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportDeviceTelemetryMsgRateLimit); + this.transportDeviceTelemetryDataPointsRateLimit = RateLimitUtil.deduplicateByDuration(transportDeviceTelemetryDataPointsRateLimit); + + this.transportGatewayMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayMsgRateLimit); + this.transportGatewayTelemetryMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayTelemetryMsgRateLimit); + this.transportGatewayTelemetryDataPointsRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayTelemetryDataPointsRateLimit); + + this.transportGatewayDeviceMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayDeviceMsgRateLimit); + this.transportGatewayDeviceTelemetryMsgRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayDeviceTelemetryMsgRateLimit); + this.transportGatewayDeviceTelemetryDataPointsRateLimit = RateLimitUtil.deduplicateByDuration(transportGatewayDeviceTelemetryDataPointsRateLimit); + + this.tenantEntityExportRateLimit = RateLimitUtil.deduplicateByDuration(tenantEntityExportRateLimit); + this.tenantEntityImportRateLimit = RateLimitUtil.deduplicateByDuration(tenantEntityImportRateLimit); + this.tenantNotificationRequestsRateLimit = RateLimitUtil.deduplicateByDuration(tenantNotificationRequestsRateLimit); + this.tenantNotificationRequestsPerRuleRateLimit = RateLimitUtil.deduplicateByDuration(tenantNotificationRequestsPerRuleRateLimit); + + this.cassandraReadQueryTenantCoreRateLimits = RateLimitUtil.deduplicateByDuration(cassandraReadQueryTenantCoreRateLimits); + this.cassandraWriteQueryTenantCoreRateLimits = RateLimitUtil.deduplicateByDuration(cassandraWriteQueryTenantCoreRateLimits); + this.cassandraReadQueryTenantRuleEngineRateLimits = RateLimitUtil.deduplicateByDuration(cassandraReadQueryTenantRuleEngineRateLimits); + this.cassandraWriteQueryTenantRuleEngineRateLimits = RateLimitUtil.deduplicateByDuration(cassandraWriteQueryTenantRuleEngineRateLimits); + + this.edgeEventRateLimits = RateLimitUtil.deduplicateByDuration(edgeEventRateLimits); + this.edgeEventRateLimitsPerEdge = RateLimitUtil.deduplicateByDuration(edgeEventRateLimitsPerEdge); + this.edgeUplinkMessagesRateLimits = RateLimitUtil.deduplicateByDuration(edgeUplinkMessagesRateLimits); + this.edgeUplinkMessagesRateLimitsPerEdge = RateLimitUtil.deduplicateByDuration(edgeUplinkMessagesRateLimitsPerEdge); + + this.wsUpdatesPerSessionRateLimit = RateLimitUtil.deduplicateByDuration(wsUpdatesPerSessionRateLimit); + + this.tenantServerRestLimitsConfiguration = RateLimitUtil.deduplicateByDuration(tenantServerRestLimitsConfiguration); + this.customerServerRestLimitsConfiguration = RateLimitUtil.deduplicateByDuration(customerServerRestLimitsConfiguration); + } + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/validation/RateLimit.java b/common/data/src/main/java/org/thingsboard/server/common/data/validation/RateLimit.java new file mode 100644 index 0000000000..9afe71da2f --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/validation/RateLimit.java @@ -0,0 +1,39 @@ +/** + * 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.validation; + +import jakarta.validation.Constraint; +import jakarta.validation.Payload; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Retention(RetentionPolicy.RUNTIME) +@Target({ElementType.FIELD, ElementType.METHOD}) +@Constraint(validatedBy = {}) +public @interface RateLimit { + + String message() default "rate limit has duplicate 'Per seconds' configuration."; + + String fieldName() default ""; + + Class[] groups() default {}; + + Class[] payload() default {}; + +} 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..6c5e62e456 --- /dev/null +++ b/common/data/src/test/java/org/thingsboard/server/common/data/limit/LimitedApiTest.java @@ -0,0 +1,101 @@ +/** + * 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); + } + } + +} diff --git a/common/data/src/test/java/org/thingsboard/server/common/data/limit/RateLimitUtilTest.java b/common/data/src/test/java/org/thingsboard/server/common/data/limit/RateLimitUtilTest.java new file mode 100644 index 0000000000..a410785126 --- /dev/null +++ b/common/data/src/test/java/org/thingsboard/server/common/data/limit/RateLimitUtilTest.java @@ -0,0 +1,99 @@ +/** + * 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.DisplayName; +import org.junit.jupiter.api.Test; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; + +import java.util.List; +import java.util.function.Function; + +import static org.assertj.core.api.Assertions.assertThat; + +class RateLimitUtilTest { + + @Test + @DisplayName("LimitedApiUtil should parse single entry correctly") + void testParseSingleEntry() { + List entries = RateLimitUtil.parseConfig("100:60"); + + assertThat(entries).hasSize(1); + assertThat(entries.get(0).capacity()).isEqualTo(100); + assertThat(entries.get(0).durationSeconds()).isEqualTo(60); + } + + @Test + @DisplayName("LimitedApiUtil should parse multiple entries correctly") + void testParseMultipleEntries() { + List entries = RateLimitUtil.parseConfig("100:60,200:30"); + + assertThat(entries).hasSize(2); + assertThat(entries.get(0).capacity()).isEqualTo(100); + assertThat(entries.get(0).durationSeconds()).isEqualTo(60); + assertThat(entries.get(1).capacity()).isEqualTo(200); + assertThat(entries.get(1).durationSeconds()).isEqualTo(30); + } + + @Test + @DisplayName("LimitedApiUtil should return empty list for null or empty config") + void testParseEmptyConfig() { + assertThat(RateLimitUtil.parseConfig(null)).isEmpty(); + assertThat(RateLimitUtil.parseConfig("")).isEmpty(); + } + + @Test + @DisplayName("LimitedApiUtil should merge two configs by summing capacities with same durations") + void testMergeStrConfigs() { + Function extractor1 = cfg -> "100:60,50:30"; + Function extractor2 = cfg -> "200:60,25:10"; + + // Fake config instance (not used directly in lambda logic) + DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); + + 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"); + } + + @Test + @DisplayName("LimitedApiUtil should merge configs when one is empty") + void testMergeWithEmptyOne() { + Function extractor1 = cfg -> "100:60"; + Function extractor2 = cfg -> ""; + + // Fake config instance (not used directly in lambda logic) + DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); + String result = RateLimitUtil.merge(extractor1, extractor2).apply(config); + + assertThat(result).isEqualTo("100:60"); + } + + @Test + @DisplayName("LimitedApiUtil should merge configs when both have distinct durations") + void testMergeWithDistinctDurations() { + Function extractor1 = cfg -> "100:60"; + Function extractor2 = cfg -> "200:10"; + + // Fake config instance (not used directly in lambda logic) + DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); + String result = RateLimitUtil.merge(extractor1, extractor2).apply(config); + + assertThat(result).isEqualTo("200:10,100:60"); + } + +} diff --git a/common/discovery-api/pom.xml b/common/discovery-api/pom.xml new file mode 100644 index 0000000000..5852761c4a --- /dev/null +++ b/common/discovery-api/pom.xml @@ -0,0 +1,73 @@ + + + 4.0.0 + + org.thingsboard + 4.1.0-SNAPSHOT + common + + org.thingsboard.common + discovery-api + jar + + Thingsboard Server Service Info Provider API + https://thingsboard.io + + + UTF-8 + ${basedir}/../.. + + + + + org.thingsboard.common + proto + + + org.thingsboard.common + message + + + + + + + org.apache.maven.plugins + maven-source-plugin + + + attach-sources + + jar + + + + + + org.apache.maven.plugins + maven-deploy-plugin + + false + + + + + + diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java b/common/discovery-api/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java similarity index 97% rename from common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java rename to common/discovery-api/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java index d7182ca2eb..b01eac8d8b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java +++ b/common/discovery-api/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java @@ -29,6 +29,8 @@ public interface TbServiceInfoProvider { ServiceInfo getServiceInfo(); + boolean isMonolith(); + boolean isService(ServiceType serviceType); ServiceInfo generateNewServiceInfoWithCurrentSystemInfo(); diff --git a/common/edqs/pom.xml b/common/edqs/pom.xml index 09181e796e..e349ae3087 100644 --- a/common/edqs/pom.xml +++ b/common/edqs/pom.xml @@ -68,6 +68,10 @@ org.thingsboard.common queue + + org.thingsboard.common + discovery-api + org.springframework.boot spring-boot-starter-web 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 8e8933c324..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 @@ -16,13 +16,16 @@ package org.thingsboard.server.common.msg.tools; import io.github.bucket4j.Bandwidth; +import io.github.bucket4j.BandwidthBuilder; import io.github.bucket4j.Bucket; -import io.github.bucket4j.Refill; import io.github.bucket4j.local.LocalBucket; import io.github.bucket4j.local.LocalBucketBuilder; import lombok.Getter; +import org.thingsboard.server.common.data.limit.RateLimitEntry; +import org.thingsboard.server.common.data.limit.RateLimitUtil; import java.time.Duration; +import java.util.List; /** * Created by ashvayka on 22.10.18. @@ -38,20 +41,19 @@ public class TbRateLimits { } public TbRateLimits(String limitsConfiguration, boolean refillIntervally) { - LocalBucketBuilder builder = Bucket.builder(); - boolean initialized = false; - for (String limitSrc : limitsConfiguration.split(",")) { - long capacity = Long.parseLong(limitSrc.split(":")[0]); - long duration = Long.parseLong(limitSrc.split(":")[1]); - Refill refill = refillIntervally ? Refill.intervally(capacity, Duration.ofSeconds(duration)) : Refill.greedy(capacity, Duration.ofSeconds(duration)); - builder.addLimit(Bandwidth.classic(capacity, refill)); - initialized = true; - } - if (initialized) { - bucket = builder.build(); - } else { + List limitedApiEntries = RateLimitUtil.parseConfig(limitsConfiguration); + if (limitedApiEntries.isEmpty()) { throw new IllegalArgumentException("Failed to parse rate limits configuration: " + limitsConfiguration); } + LocalBucketBuilder localBucket = Bucket.builder(); + for (RateLimitEntry entry : limitedApiEntries) { + BandwidthBuilder.BandwidthBuilderRefillStage bandwidthBuilder = Bandwidth.builder().capacity(entry.capacity()); + Bandwidth bandwidth = refillIntervally ? + bandwidthBuilder.refillIntervally(entry.capacity(), Duration.ofSeconds(entry.durationSeconds())).build() : + bandwidthBuilder.refillGreedy(entry.capacity(), Duration.ofSeconds(entry.durationSeconds())).build(); + localBucket.addLimit(bandwidth); + } + this.bucket = localBucket.build(); this.configuration = limitsConfiguration; } diff --git a/common/message/src/test/java/org/thingsboard/server/common/msg/tools/RateLimitsTest.java b/common/message/src/test/java/org/thingsboard/server/common/msg/tools/TbRateLimitsTest.java similarity index 67% rename from common/message/src/test/java/org/thingsboard/server/common/msg/tools/RateLimitsTest.java rename to common/message/src/test/java/org/thingsboard/server/common/msg/tools/TbRateLimitsTest.java index 9d27687cd9..27a2fe6286 100644 --- a/common/message/src/test/java/org/thingsboard/server/common/msg/tools/RateLimitsTest.java +++ b/common/message/src/test/java/org/thingsboard/server/common/msg/tools/TbRateLimitsTest.java @@ -16,14 +16,16 @@ package org.thingsboard.server.common.msg.tools; import org.awaitility.pollinterval.FixedPollInterval; +import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.awaitility.Awaitility.await; -public class RateLimitsTest { +public class TbRateLimitsTest { @Test public void testRateLimits_greedyRefill() { @@ -83,4 +85,41 @@ public class RateLimitsTest { }); } + @Test + @DisplayName("TbRateLimits should construct with single rate limit") + void testSingleLimitConstructor() { + TbRateLimits limits = new TbRateLimits("10:1", false); + assertThat(limits.getConfiguration()).isEqualTo("10:1"); + } + + @Test + @DisplayName("TbRateLimits should construct with multiple rate limits") + void testMultipleLimitConstructor() { + String config = "10:1,100:10"; + TbRateLimits limits = new TbRateLimits(config, false); + assertThat(limits.getConfiguration()).isEqualTo(config); + } + + @Test + @DisplayName("TbRateLimits should throw IllegalArgumentException on empty string") + void testEmptyConfigThrows() { + assertThatThrownBy(() -> new TbRateLimits("", false)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Failed to parse rate limits configuration: "); + } + + @Test + @DisplayName("TbRateLimits should throw NumberFormatException on malformed value") + void testMalformedConfigThrows() { + assertThatThrownBy(() -> new TbRateLimits("not_a_number:second", false)) + .isInstanceOf(NumberFormatException.class); + } + + @Test + @DisplayName("TbRateLimits should throw ArrayIndexOutOfBoundsException on missing colon") + void testColonMissingThrows() { + assertThatThrownBy(() -> new TbRateLimits("100", false)) + .isInstanceOf(ArrayIndexOutOfBoundsException.class); + } + } diff --git a/common/pom.xml b/common/pom.xml index 6afae4e378..0915f47f1a 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -50,6 +50,7 @@ version-control script edqs + discovery-api diff --git a/common/queue/pom.xml b/common/queue/pom.xml index 2e4b14e282..f2f3091999 100644 --- a/common/queue/pom.xml +++ b/common/queue/pom.xml @@ -60,6 +60,10 @@ org.thingsboard.common cluster-api + + org.thingsboard.common + discovery-api + org.apache.kafka kafka-clients diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java index 3c0b6f5b51..5f23f35070 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java @@ -91,11 +91,9 @@ public class DefaultTbServiceInfoProvider implements TbServiceInfoProvider { } } log.info("Current Service ID: {}", serviceId); - if (serviceType.equalsIgnoreCase("monolith")) { - serviceTypes = List.of(ServiceType.values()); - } else { - serviceTypes = Collections.singletonList(ServiceType.of(serviceType)); - } + serviceTypes = isMonolith() ? + List.of(ServiceType.values()) : + Collections.singletonList(ServiceType.of(serviceType)); if (!serviceTypes.contains(ServiceType.TB_RULE_ENGINE) || assignedTenantProfiles == null) { assignedTenantProfiles = Collections.emptySet(); } @@ -134,6 +132,11 @@ public class DefaultTbServiceInfoProvider implements TbServiceInfoProvider { return serviceInfo; } + @Override + public boolean isMonolith() { + return serviceType.equalsIgnoreCase("monolith"); + } + @Override public boolean isService(ServiceType serviceType) { return serviceTypes.contains(serviceType); diff --git a/common/transport/transport-api/pom.xml b/common/transport/transport-api/pom.xml index 2cd41064da..7f54dcbc85 100644 --- a/common/transport/transport-api/pom.xml +++ b/common/transport/transport-api/pom.xml @@ -60,6 +60,10 @@ org.thingsboard.common util + + org.thingsboard.common + discovery-api + com.google.code.gson gson diff --git a/dao/pom.xml b/dao/pom.xml index 2206d8a7d7..9f411f8302 100644 --- a/dao/pom.xml +++ b/dao/pom.xml @@ -59,6 +59,10 @@ org.thingsboard.common util + + org.thingsboard.common + discovery-api + com.networknt json-schema-validator 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 259cd908fc..7befb1db62 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 @@ -28,7 +28,9 @@ import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.dao.entity.EntityService; import org.thingsboard.server.dao.util.AbstractBufferedRateExecutor; import org.thingsboard.server.dao.util.AsyncTaskContext; +import org.thingsboard.server.dao.util.BufferedRateExecutorType; import org.thingsboard.server.dao.util.NoSqlAnyDao; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; /** * Created by ashvayka on 24.10.18. @@ -38,8 +40,6 @@ import org.thingsboard.server.dao.util.NoSqlAnyDao; @NoSqlAnyDao public class CassandraBufferedRateReadExecutor extends AbstractBufferedRateExecutor { - static final String BUFFER_NAME = "Read"; - public CassandraBufferedRateReadExecutor( @Value("${cassandra.query.buffer_size}") int queueLimit, @Value("${cassandra.query.concurrent_limit}") int concurrencyLimit, @@ -51,9 +51,10 @@ public class CassandraBufferedRateReadExecutor extends AbstractBufferedRateExecu @Value("${cassandra.query.print_queries_freq:0}") int printQueriesFreq, @Autowired StatsFactory statsFactory, @Autowired EntityService entityService, - @Autowired RateLimitService rateLimitService) { - super(queueLimit, concurrencyLimit, maxWaitTime, dispatcherThreads, callbackThreads, pollMs, printQueriesFreq, statsFactory, - entityService, rateLimitService, printTenantNames); + @Autowired RateLimitService rateLimitService, + @Autowired(required = false) TbServiceInfoProvider serviceInfoProvider) { + super(queueLimit, concurrencyLimit, maxWaitTime, dispatcherThreads, callbackThreads, pollMs, printQueriesFreq, + BufferedRateExecutorType.READ, serviceInfoProvider, rateLimitService, statsFactory, entityService, printTenantNames); } @Scheduled(fixedDelayString = "${cassandra.query.rate_limit_print_interval_ms}") @@ -67,11 +68,6 @@ public class CassandraBufferedRateReadExecutor extends AbstractBufferedRateExecu super.stop(); } - @Override - public String getBufferName() { - return BUFFER_NAME; - } - @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 9c894a7eac..9b7db4fae4 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 @@ -28,7 +28,9 @@ import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.dao.entity.EntityService; import org.thingsboard.server.dao.util.AbstractBufferedRateExecutor; import org.thingsboard.server.dao.util.AsyncTaskContext; +import org.thingsboard.server.dao.util.BufferedRateExecutorType; import org.thingsboard.server.dao.util.NoSqlAnyDao; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; /** * Created by ashvayka on 24.10.18. @@ -38,8 +40,6 @@ import org.thingsboard.server.dao.util.NoSqlAnyDao; @NoSqlAnyDao public class CassandraBufferedRateWriteExecutor extends AbstractBufferedRateExecutor { - static final String BUFFER_NAME = "Write"; - public CassandraBufferedRateWriteExecutor( @Value("${cassandra.query.buffer_size}") int queueLimit, @Value("${cassandra.query.concurrent_limit}") int concurrencyLimit, @@ -51,9 +51,10 @@ public class CassandraBufferedRateWriteExecutor extends AbstractBufferedRateExec @Value("${cassandra.query.print_queries_freq:0}") int printQueriesFreq, @Autowired StatsFactory statsFactory, @Autowired EntityService entityService, - @Autowired RateLimitService rateLimitService) { - super(queueLimit, concurrencyLimit, maxWaitTime, dispatcherThreads, callbackThreads, pollMs, printQueriesFreq, statsFactory, - entityService, rateLimitService, printTenantNames); + @Autowired RateLimitService rateLimitService, + @Autowired(required = false) TbServiceInfoProvider serviceInfoProvider) { + super(queueLimit, concurrencyLimit, maxWaitTime, dispatcherThreads, callbackThreads, pollMs, printQueriesFreq, + BufferedRateExecutorType.WRITE, serviceInfoProvider, rateLimitService, statsFactory, entityService, printTenantNames); } @Scheduled(fixedDelayString = "${cassandra.query.rate_limit_print_interval_ms}") @@ -67,11 +68,6 @@ public class CassandraBufferedRateWriteExecutor extends AbstractBufferedRateExec super.stop(); } - @Override - public String getBufferName() { - return BUFFER_NAME; - } - @Override protected SettableFuture create() { return SettableFuture.create(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/ConstraintValidator.java b/dao/src/main/java/org/thingsboard/server/dao/service/ConstraintValidator.java index 4992af5b96..f836f229ea 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/service/ConstraintValidator.java +++ b/dao/src/main/java/org/thingsboard/server/dao/service/ConstraintValidator.java @@ -33,6 +33,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.validation.beanvalidation.LocalValidatorFactoryBean; import org.thingsboard.server.common.data.validation.Length; import org.thingsboard.server.common.data.validation.NoXss; +import org.thingsboard.server.common.data.validation.RateLimit; import org.thingsboard.server.dao.exception.DataValidationException; import java.util.Collection; @@ -103,6 +104,7 @@ public class ConstraintValidator { ConstraintMapping constraintMapping = new DefaultConstraintMapping(null); constraintMapping.constraintDefinition(NoXss.class).validatedBy(NoXssValidator.class); constraintMapping.constraintDefinition(Length.class).validatedBy(StringLengthValidator.class); + constraintMapping.constraintDefinition(RateLimit.class).validatedBy(RateLimitValidator.class); return constraintMapping; } 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 new file mode 100644 index 0000000000..bf776d4dc7 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/service/RateLimitValidator.java @@ -0,0 +1,32 @@ +/** + * 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.dao.service; + +import jakarta.validation.ConstraintValidator; +import jakarta.validation.ConstraintValidatorContext; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.data.limit.RateLimitUtil; +import org.thingsboard.server.common.data.validation.RateLimit; + +@Slf4j +public class RateLimitValidator implements ConstraintValidator { + + @Override + public boolean isValid(String value, ConstraintValidatorContext constraintValidatorContext) { + return value == null || RateLimitUtil.isValid(value); + } + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/validator/TenantProfileDataValidator.java b/dao/src/main/java/org/thingsboard/server/dao/service/validator/TenantProfileDataValidator.java index fae3ab43d6..6f6b6a3528 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/service/validator/TenantProfileDataValidator.java +++ b/dao/src/main/java/org/thingsboard/server/dao/service/validator/TenantProfileDataValidator.java @@ -24,6 +24,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.ProcessingStrategy; import org.thingsboard.server.common.data.queue.SubmitStrategy; 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.TenantProfileQueueConfiguration; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.service.DataValidator; @@ -51,9 +52,12 @@ public class TenantProfileDataValidator extends DataValidator { if (tenantProfile.getProfileData() == null) { throw new DataValidationException("Tenant profile data should be specified!"); } - if (tenantProfile.getProfileData().getConfiguration() == null) { + + Optional profileConfiguration = tenantProfile.getProfileConfiguration(); + if (profileConfiguration.isEmpty()) { throw new DataValidationException("Tenant profile data configuration should be specified!"); } + if (tenantProfile.isDefault()) { TenantProfile defaultTenantProfile = tenantProfileService.findDefaultTenantProfile(tenantId); if (defaultTenantProfile != null && !defaultTenantProfile.getId().equals(tenantProfile.getId())) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java b/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java index a87e9312e3..cbcf3e81ec 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java +++ b/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java @@ -34,12 +34,14 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.cache.limits.RateLimitService; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.limit.LimitedApi; +import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.stats.DefaultCounter; import org.thingsboard.server.common.stats.StatsCounter; import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.common.stats.StatsType; import org.thingsboard.server.dao.entity.EntityService; import org.thingsboard.server.dao.nosql.CassandraStatementTask; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import java.util.HashMap; import java.util.Map; @@ -64,6 +66,7 @@ public abstract class AbstractBufferedRateExecutor> queue; private final ExecutorService dispatcherExecutor; private final ExecutorService callbackExecutor; @@ -82,19 +85,23 @@ 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, boolean printTenantNames) { + int callbackThreads, long pollMs, int printQueriesFreq, BufferedRateExecutorType executorType, TbServiceInfoProvider serviceInfoProvider, + RateLimitService rateLimitService, StatsFactory statsFactory, EntityService entityService, boolean printTenantNames) { this.maxWaitTime = maxWaitTime; this.pollMs = pollMs; + this.bufferName = executorType.getDisplayName(); + this.myLimitedApi = resolveLimitedApi(serviceInfoProvider, executorType); 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; @@ -114,14 +121,14 @@ public abstract class AbstractBufferedRateExecutor execute(AsyncTaskContext taskCtx); - public abstract String getBufferName(); - 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; @@ -169,7 +184,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); @@ -222,7 +237,7 @@ public abstract class AbstractBufferedRateExecutor taskCtx) { @@ -298,7 +313,7 @@ public abstract class AbstractBufferedRateExecutorcluster-api ${project.version} + + org.thingsboard.common + discovery-api + ${project.version} + org.thingsboard.rule-engine rule-engine-api diff --git a/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.html b/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.html index aded3e2f78..3c5c8774d5 100644 --- a/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.html +++ b/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.html @@ -693,11 +693,19 @@
- + + + + +
+
+ - +
@@ -725,8 +733,8 @@
- +
diff --git a/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.ts b/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.ts index b1d6652e4a..c6efd9dee3 100644 --- a/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.ts +++ b/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.ts @@ -114,7 +114,10 @@ export class DefaultTenantProfileConfigurationComponent implements ControlValueA maxWsSubscriptionsPerRegularUser: [null, [Validators.min(0)]], maxWsSubscriptionsPerPublicUser: [null, [Validators.min(0)]], wsUpdatesPerSessionRateLimit: [null, []], - cassandraQueryTenantRateLimitsConfiguration: [null, []], + cassandraWriteQueryTenantCoreRateLimits: [null, []], + cassandraReadQueryTenantCoreRateLimits: [null, []], + cassandraWriteQueryTenantRuleEngineRateLimits: [null, []], + cassandraReadQueryTenantRuleEngineRateLimits: [null, []], edgeEventRateLimits: [null, []], edgeEventRateLimitsPerEdge: [null, []], edgeUplinkMessagesRateLimits: [null, []], diff --git a/ui-ngx/src/app/modules/home/components/profile/tenant/rate-limits/rate-limits.models.ts b/ui-ngx/src/app/modules/home/components/profile/tenant/rate-limits/rate-limits.models.ts index a4ea52cf62..898a3baaf7 100644 --- a/ui-ngx/src/app/modules/home/components/profile/tenant/rate-limits/rate-limits.models.ts +++ b/ui-ngx/src/app/modules/home/components/profile/tenant/rate-limits/rate-limits.models.ts @@ -37,7 +37,10 @@ export enum RateLimitsType { TENANT_SERVER_REST_LIMITS_CONFIGURATION = 'TENANT_SERVER_REST_LIMITS_CONFIGURATION', CUSTOMER_SERVER_REST_LIMITS_CONFIGURATION = 'CUSTOMER_SERVER_REST_LIMITS_CONFIGURATION', WS_UPDATE_PER_SESSION_RATE_LIMIT = 'WS_UPDATE_PER_SESSION_RATE_LIMIT', - CASSANDRA_QUERY_TENANT_RATE_LIMITS_CONFIGURATION = 'CASSANDRA_QUERY_TENANT_RATE_LIMITS_CONFIGURATION', + CASSANDRA_WRITE_QUERY_TENANT_CORE_RATE_LIMITS = 'CASSANDRA_WRITE_QUERY_TENANT_CORE_RATE_LIMITS', + CASSANDRA_READ_QUERY_TENANT_CORE_RATE_LIMITS = 'CASSANDRA_READ_QUERY_TENANT_CORE_RATE_LIMITS', + CASSANDRA_WRITE_QUERY_TENANT_RULE_ENGINE_RATE_LIMITS = 'CASSANDRA_WRITE_QUERY_TENANT_RULE_ENGINE_RATE_LIMITS', + CASSANDRA_READ_QUERY_TENANT_RULE_ENGINE_RATE_LIMITS = 'CASSANDRA_READ_QUERY_TENANT_RULE_ENGINE_RATE_LIMITS', TENANT_ENTITY_EXPORT_RATE_LIMIT = 'TENANT_ENTITY_EXPORT_RATE_LIMIT', TENANT_ENTITY_IMPORT_RATE_LIMIT = 'TENANT_ENTITY_IMPORT_RATE_LIMIT', TENANT_NOTIFICATION_REQUEST_RATE_LIMIT = 'TENANT_NOTIFICATION_REQUEST_RATE_LIMIT', @@ -66,7 +69,10 @@ export const rateLimitsLabelTranslationMap = new Map( [RateLimitsType.TENANT_SERVER_REST_LIMITS_CONFIGURATION, 'tenant-profile.rest-requests-for-tenant'], [RateLimitsType.CUSTOMER_SERVER_REST_LIMITS_CONFIGURATION, 'tenant-profile.customer-rest-limits'], [RateLimitsType.WS_UPDATE_PER_SESSION_RATE_LIMIT, 'tenant-profile.ws-limit-updates-per-session'], - [RateLimitsType.CASSANDRA_QUERY_TENANT_RATE_LIMITS_CONFIGURATION, 'tenant-profile.cassandra-tenant-limits-configuration'], + [RateLimitsType.CASSANDRA_WRITE_QUERY_TENANT_CORE_RATE_LIMITS, 'tenant-profile.cassandra-write-tenant-core-limits-configuration'], + [RateLimitsType.CASSANDRA_READ_QUERY_TENANT_CORE_RATE_LIMITS, 'tenant-profile.cassandra-read-tenant-core-limits-configuration'], + [RateLimitsType.CASSANDRA_WRITE_QUERY_TENANT_RULE_ENGINE_RATE_LIMITS, 'tenant-profile.cassandra-write-tenant-rule-engine-limits-configuration'], + [RateLimitsType.CASSANDRA_READ_QUERY_TENANT_RULE_ENGINE_RATE_LIMITS, 'tenant-profile.cassandra-read-tenant-rule-engine-limits-configuration'], [RateLimitsType.TENANT_ENTITY_EXPORT_RATE_LIMIT, 'tenant-profile.tenant-entity-export-rate-limit'], [RateLimitsType.TENANT_ENTITY_IMPORT_RATE_LIMIT, 'tenant-profile.tenant-entity-import-rate-limit'], [RateLimitsType.TENANT_NOTIFICATION_REQUEST_RATE_LIMIT, 'tenant-profile.tenant-notification-request-rate-limit'], @@ -96,7 +102,10 @@ export const rateLimitsDialogTitleTranslationMap = new Map