diff --git a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java index 2f157af70e..f5b769e568 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java @@ -19,6 +19,7 @@ import lombok.Data; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Component; +import org.thingsboard.server.cache.limits.RateLimitService; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; import org.thingsboard.server.dao.asset.AssetProfileService; @@ -150,6 +151,9 @@ public class EdgeContextComponent { @Autowired private ResourceService resourceService; + @Autowired + private RateLimitService rateLimitService; + @Autowired private NotificationRuleProcessor notificationRuleProcessor; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 4924827e35..2580d399af 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.StringDataEntry; +import org.thingsboard.server.common.data.limit.LimitedApi; import org.thingsboard.server.common.data.notification.rule.trigger.EdgeCommunicationFailureTrigger; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; @@ -109,6 +110,7 @@ public final class EdgeGrpcSession implements Closeable { private static final String QUEUE_START_TS_ATTR_KEY = "queueStartTs"; private static final String QUEUE_START_SEQ_ID_ATTR_KEY = "queueStartSeqId"; + private static final String RATE_LIMIT_REACHED = "Rate limit reached"; private final UUID sessionId; private final BiConsumer sessionOpenListener; @@ -264,36 +266,51 @@ public final class EdgeGrpcSession implements Closeable { } private void onUplinkMsg(UplinkMsg uplinkMsg) { + if (isRateLimitViolated(uplinkMsg)) { + return; + } ListenableFuture> future = processUplinkMsg(uplinkMsg); Futures.addCallback(future, new FutureCallback<>() { @Override public void onSuccess(@Nullable List result) { - UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder() - .setUplinkMsgId(uplinkMsg.getUplinkMsgId()) - .setSuccess(true).build(); - sendDownlinkMsg(ResponseMsg.newBuilder() - .setUplinkResponseMsg(uplinkResponseMsg) - .build()); + sendResponseMessage(uplinkMsg.getUplinkMsgId(), true, null); } @Override public void onFailure(Throwable t) { String errorMsg = EdgeUtils.createErrorMsgFromRootCauseAndStackTrace(t); - UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder() - .setUplinkMsgId(uplinkMsg.getUplinkMsgId()) - .setSuccess(false).setErrorMsg(errorMsg).build(); - sendDownlinkMsg(ResponseMsg.newBuilder() - .setUplinkResponseMsg(uplinkResponseMsg) - .build()); + sendResponseMessage(uplinkMsg.getUplinkMsgId(), false, errorMsg); } }, ctx.getGrpcCallbackExecutorService()); } + private boolean isRateLimitViolated(UplinkMsg uplinkMsg) { + if (!ctx.getRateLimitService().checkRateLimit(LimitedApi.EDGE_UPLINK_MESSAGES, tenantId) || + !ctx.getRateLimitService().checkRateLimit(LimitedApi.EDGE_UPLINK_MESSAGES_PER_EDGE, tenantId, edge.getId())) { + String errorMsg = String.format("Failed to process uplink message. %s", RATE_LIMIT_REACHED); + sendResponseMessage(uplinkMsg.getUplinkMsgId(), false, errorMsg); + return true; + } + return false; + } + + private void sendResponseMessage(int uplinkMsgId, boolean success, String errorMsg) { + UplinkResponseMsg.Builder responseBuilder = UplinkResponseMsg.newBuilder() + .setUplinkMsgId(uplinkMsgId) + .setSuccess(success); + if (errorMsg != null) { + responseBuilder.setErrorMsg(errorMsg); + } + sendDownlinkMsg(ResponseMsg.newBuilder() + .setUplinkResponseMsg(responseBuilder.build()) + .build()); + } + private void onDownlinkResponse(DownlinkResponseMsg msg) { try { if (msg.getSuccess()) { sessionState.getPendingMsgsMap().remove(msg.getDownlinkMsgId()); - log.debug("[{}][{}] Msg has been processed successfully!Msg Id: [{}], Msg: {}", this.tenantId, edge.getRoutingKey(), msg.getDownlinkMsgId(), msg); + log.debug("[{}][{}] Msg has been processed successfully! Msg Id: [{}], Msg: {}", this.tenantId, edge.getRoutingKey(), msg.getDownlinkMsgId(), msg); } else { log.error("[{}][{}] Msg processing failed! Msg Id: [{}], Error msg: {}", this.tenantId, edge.getRoutingKey(), msg.getDownlinkMsgId(), msg.getErrorMsg()); } 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 2570b80f9f..95a2c63f52 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 @@ -71,6 +71,8 @@ public class RateLimitServiceTest { profileConfiguration.setCassandraQueryTenantRateLimitsConfiguration(rateLimit); profileConfiguration.setEdgeEventRateLimits(rateLimit); profileConfiguration.setEdgeEventRateLimitsPerEdge(rateLimit); + profileConfiguration.setEdgeUplinkMessagesRateLimits(rateLimit); + profileConfiguration.setEdgeUplinkMessagesRateLimitsPerEdge(rateLimit); updateTenantProfileConfiguration(profileConfiguration); for (LimitedApi limitedApi : List.of( @@ -80,7 +82,9 @@ public class RateLimitServiceTest { LimitedApi.REST_REQUESTS_PER_CUSTOMER, LimitedApi.CASSANDRA_QUERIES, LimitedApi.EDGE_EVENTS, - LimitedApi.EDGE_EVENTS_PER_EDGE + LimitedApi.EDGE_EVENTS_PER_EDGE, + LimitedApi.EDGE_UPLINK_MESSAGES, + LimitedApi.EDGE_UPLINK_MESSAGES_PER_EDGE )) { testRateLimits(limitedApi, max, tenantId); } 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 980eb880a0..1a0f038c90 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 @@ -33,6 +33,8 @@ public enum LimitedApi { CASSANDRA_QUERIES(DefaultTenantProfileConfiguration::getCassandraQueryTenantRateLimitsConfiguration, "Cassandra 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), + EDGE_UPLINK_MESSAGES_PER_EDGE(DefaultTenantProfileConfiguration::getEdgeUplinkMessagesRateLimitsPerEdge, "Edge uplink messages per edge", false), PASSWORD_RESET(false, true), TWO_FA_VERIFICATION_CODE_SEND(false, true), TWO_FA_VERIFICATION_CODE_CHECK(false, true), 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 3fe110de0b..a0cfbfe1e1 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 @@ -83,6 +83,8 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura private String edgeEventRateLimits; private String edgeEventRateLimitsPerEdge; + private String edgeUplinkMessagesRateLimits; + private String edgeUplinkMessagesRateLimitsPerEdge; private int defaultStorageTtlDays; private int alarmsTtlDays; 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 867cd74f3b..ef024fe0a2 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 @@ -571,6 +571,14 @@ [type]="rateLimitsType.EDGE_EVENTS_PER_EDGE_RATE_LIMIT"> +
+ + + + +
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 0c95324d85..c13eed9c03 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 @@ -108,7 +108,9 @@ export class DefaultTenantProfileConfigurationComponent implements ControlValueA wsUpdatesPerSessionRateLimit: [null, []], cassandraQueryTenantRateLimitsConfiguration: [null, []], edgeEventRateLimits: [null, []], - edgeEventRateLimitsPerEdge: [null, []] + edgeEventRateLimitsPerEdge: [null, []], + edgeUplinkMessagesRateLimits: [null, []], + edgeUplinkMessagesRateLimitsPerEdge: [null, []] }); this.defaultTenantProfileConfigurationFormGroup.get('smsEnabled').valueChanges.pipe( 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 8924682aa1..ee13ea9b23 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,9 @@ export enum RateLimitsType { TENANT_NOTIFICATION_REQUEST_RATE_LIMIT = 'TENANT_NOTIFICATION_REQUEST_RATE_LIMIT', TENANT_NOTIFICATION_REQUESTS_PER_RULE_RATE_LIMIT = 'TENANT_NOTIFICATION_REQUESTS_PER_RULE_RATE_LIMIT', EDGE_EVENTS_RATE_LIMIT = 'EDGE_EVENTS_RATE_LIMIT', - EDGE_EVENTS_PER_EDGE_RATE_LIMIT = 'EDGE_EVENTS_PER_EDGE_RATE_LIMIT' + EDGE_EVENTS_PER_EDGE_RATE_LIMIT = 'EDGE_EVENTS_PER_EDGE_RATE_LIMIT', + EDGE_UPLINK_MESSAGES_RATE_LIMIT = 'EDGE_UPLINK_MESSAGES_RATE_LIMIT', + EDGE_UPLINK_MESSAGES_PER_EDGE_RATE_LIMIT = 'EDGE_UPLINK_MESSAGES_PER_EDGE_RATE_LIMIT' } export const rateLimitsLabelTranslationMap = new Map( @@ -58,6 +60,8 @@ export const rateLimitsLabelTranslationMap = new Map( [RateLimitsType.TENANT_NOTIFICATION_REQUESTS_PER_RULE_RATE_LIMIT, 'tenant-profile.tenant-notification-requests-per-rule-rate-limit'], [RateLimitsType.EDGE_EVENTS_RATE_LIMIT, 'tenant-profile.rate-limits.edge-events-rate-limit'], [RateLimitsType.EDGE_EVENTS_PER_EDGE_RATE_LIMIT, 'tenant-profile.rate-limits.edge-events-per-edge-rate-limit'], + [RateLimitsType.EDGE_UPLINK_MESSAGES_RATE_LIMIT, 'tenant-profile.rate-limits.edge-uplink-messages-rate-limit'], + [RateLimitsType.EDGE_UPLINK_MESSAGES_PER_EDGE_RATE_LIMIT, 'tenant-profile.rate-limits.edge-uplink-messages-per-edge-rate-limit'], ] ); @@ -79,6 +83,8 @@ export const rateLimitsDialogTitleTranslationMap = new Map( @@ -43,5 +45,7 @@ export const LimitedApiTranslationMap = new Map( [LimitedApi.TRANSPORT_MESSAGES_PER_DEVICE, 'api-limit.transport-messages-per-device'], [LimitedApi.EDGE_EVENTS, 'api-limit.edge-events'], [LimitedApi.EDGE_EVENTS_PER_EDGE, 'api-limit.edge-events-per-edge'], + [LimitedApi.EDGE_UPLINK_MESSAGES, 'api-limit.edge-uplink-messages'], + [LimitedApi.EDGE_UPLINK_MESSAGES_PER_EDGE, 'api-limit.edge-uplink-messages-per-edge'] ] ); diff --git a/ui-ngx/src/assets/help/en_US/notification/rate_limits.md b/ui-ngx/src/assets/help/en_US/notification/rate_limits.md index d314e681c2..c17610f39a 100644 --- a/ui-ngx/src/assets/help/en_US/notification/rate_limits.md +++ b/ui-ngx/src/assets/help/en_US/notification/rate_limits.md @@ -11,7 +11,7 @@ Available template parameters: * `api` - rate-limited API label; one of: 'REST API requests', 'REST API requests per customer', 'transport messages', 'transport messages per device', 'Cassandra queries', 'WS updates per session', 'notification requests', 'notification requests per rule', - 'entity version creation', 'entity version load', 'Edge events', 'Edge events per edge'; + 'entity version creation', 'entity version load', 'Edge events', 'Edge events per edge', 'Edge uplink messages', 'Edge uplink messages per edge'; * `limitLevelEntityType` - entity type of the limit level entity, e.g. 'Tenant', 'Device', 'Notification rule', 'Customer', etc.; * `limitLevelEntityId` - id of the limit level entity; * `limitLevelEntityName` - name of the limit level entity; diff --git a/ui-ngx/src/assets/locale/locale.constant-en_US.json b/ui-ngx/src/assets/locale/locale.constant-en_US.json index 0bc6462b36..1948b84e79 100644 --- a/ui-ngx/src/assets/locale/locale.constant-en_US.json +++ b/ui-ngx/src/assets/locale/locale.constant-en_US.json @@ -861,7 +861,9 @@ "transport-messages-per-device": "Transport messages per device", "ws-updates-per-session": "WS updates per session", "edge-events": "Edge events", - "edge-events-per-edge": "Edge events per edge" + "edge-events-per-edge": "Edge events per edge", + "edge-uplink-messages": "Edge uplink messages", + "edge-uplink-messages-per-edge": "Edge uplink messages per edge" }, "audit-log": { "audit": "Audit", @@ -4210,6 +4212,10 @@ "edit-edge-events-per-edge-rate-limit": "Edit edge events per edge rate limits", "edge-events-rate-limit": "Edge events", "edge-events-per-edge-rate-limit": "Edge events per edge", + "edit-edge-uplink-messages-rate-limit": "Edit edge uplink messages rate limits", + "edit-edge-uplink-messages-per-edge-rate-limit": "Edit edge uplink messages per edge rate limits", + "edge-uplink-messages-rate-limit": "Edge uplink messages", + "edge-uplink-messages-per-edge-rate-limit": "Edge uplink messages per edge", "messages-per": "messages per", "not-set": "Not set", "number-of-messages": "Number of messages",