tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Refactor RateLimitsUpdater; handle rate limits disabling

pull/5946/head
Viacheslav Klimov 4 years ago
parent
ffd2a02742
commit
d99a8066fe
8 changed files with 74 additions and 91 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 56
      application/src/main/java/org/thingsboard/server/config/RateLimitProcessingFilter.java
  2. 16
      application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java
  3. 34
      application/src/main/java/org/thingsboard/server/service/install/update/RateLimitsUpdater.java
  4. 4
      application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java
  5. 21
      application/src/main/resources/thingsboard.yml
  6. 10
      common/data/src/main/java/org/thingsboard/server/common/data/TenantProfile.java
  7. 20
      common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java
  8. 4
      dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java

56
application/src/main/java/org/thingsboard/server/config/RateLimitProcessingFilter.java
View File

@ -15,14 +15,14 @@
*/
package org.thingsboard.server.config;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Component;
import org.springframework.web.filter.GenericFilterBean;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.tools.TbRateLimits;
import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
@ -36,6 +36,7 @@ import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -55,37 +56,38 @@ public class RateLimitProcessingFilter extends GenericFilterBean {
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
SecurityUser user = getCurrentUser();
if (user != null && !user.isSystemAdmin()) {
var profileConfiguration = tenantProfileCache.get(user.getTenantId()).getDefaultTenantProfileConfiguration();
if (profileConfiguration != null) {
if (user.isTenantAdmin() && StringUtils.isNotEmpty(profileConfiguration.getTenantServerRestLimitsConfiguration())) {
TbRateLimits rateLimits = perTenantLimits.get(user.getTenantId());
if (rateLimits == null || !rateLimits.getConfiguration().equals(profileConfiguration.getTenantServerRestLimitsConfiguration())) {
// fixme: or maybe handle component lifecycle event ?
rateLimits = new TbRateLimits(profileConfiguration.getTenantServerRestLimitsConfiguration());
perTenantLimits.put(user.getTenantId(), rateLimits);
}
if (!rateLimits.tryConsume()) {
errorResponseHandler.handle(new TbRateLimitsException(EntityType.TENANT), (HttpServletResponse) response);
return;
}
} else if (user.isCustomerUser() && StringUtils.isNotEmpty(profileConfiguration.getCustomerServerRestLimitsConfiguration())) {
TbRateLimits rateLimits = perCustomerLimits.get(user.getCustomerId());
if (rateLimits == null || !rateLimits.getConfiguration().equals(profileConfiguration.getCustomerServerRestLimitsConfiguration())) {
rateLimits = new TbRateLimits(profileConfiguration.getCustomerServerRestLimitsConfiguration());
perCustomerLimits.put(user.getCustomerId(), rateLimits);
}
if (!rateLimits.tryConsume()) {
errorResponseHandler.handle(new TbRateLimitsException(EntityType.CUSTOMER), (HttpServletResponse) response);
return;
}
var profileConfiguration = tenantProfileCache.get(user.getTenantId()).getDefaultProfileConfiguration();
if (!checkRateLimits(user.getTenantId(), profileConfiguration.getTenantServerRestLimitsConfiguration(), perTenantLimits, response)) {
return;
}
if (user.isCustomerUser()) {
if (!checkRateLimits(user.getCustomerId(), profileConfiguration.getCustomerServerRestLimitsConfiguration(), perCustomerLimits, response)) {
return;
}
}
}
chain.doFilter(request, response);
}
private <I extends EntityId> boolean checkRateLimits(I ownerId, String rateLimitConfig, Map<I, TbRateLimits> rateLimitsMap, ServletResponse response) {
if (StringUtils.isNotEmpty(rateLimitConfig)) {
TbRateLimits rateLimits = rateLimitsMap.get(ownerId);
if (rateLimits == null || !rateLimits.getConfiguration().equals(rateLimitConfig)) {
rateLimits = new TbRateLimits(rateLimitConfig);
rateLimitsMap.put(ownerId, rateLimits);
}
if (!rateLimits.tryConsume()) {
errorResponseHandler.handle(new TbRateLimitsException(ownerId.getEntityType()), (HttpServletResponse) response);
return false;
}
} else {
rateLimitsMap.remove(ownerId);
}
return true;
}
protected SecurityUser getCurrentUser() {
Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
if (authentication != null && authentication.getPrincipal() instanceof SecurityUser) {

16
application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java
View File

@ -38,7 +38,6 @@ import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.security.model.SecurityUser;
import org.thingsboard.server.service.security.model.UserPrincipal;
import org.thingsboard.server.service.telemetry.DefaultTelemetryWebSocketService;
import org.thingsboard.server.service.telemetry.SessionEvent;
import org.thingsboard.server.service.telemetry.TelemetryWebSocketMsgEndpoint;
import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
@ -138,8 +137,9 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr
if (!checkLimits(session, sessionRef)) {
return;
}
var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultTenantProfileConfiguration();
internalSessionMap.put(internalSessionId, new SessionMetaData(session, sessionRef, tenantProfileConfiguration.getWsMsgQueueLimitPerSession()));
var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration();
internalSessionMap.put(internalSessionId, new SessionMetaData(session, sessionRef, tenantProfileConfiguration.getWsMsgQueueLimitPerSession() > 0 ?
tenantProfileConfiguration.getWsMsgQueueLimitPerSession() : 500));
externalSessionMap.put(externalSessionId, internalSessionId);
processInWebSocketService(sessionRef, SessionEvent.onEstablished());
@ -300,9 +300,8 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr
if (internalId != null) {
SessionMetaData sessionMd = internalSessionMap.get(internalId);
if (sessionMd != null) {
var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultTenantProfileConfiguration();
var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration();
if (StringUtils.isNotEmpty(tenantProfileConfiguration.getWsUpdatesPerSessionRateLimit())) {
// fixme: is this ok not to update rate limits config if it was change in profile?
TbRateLimits rateLimits = perSessionUpdateLimits.computeIfAbsent(sessionRef.getSessionId(), sid -> new TbRateLimits(tenantProfileConfiguration.getWsUpdatesPerSessionRateLimit()));
if (!rateLimits.tryConsume()) {
if (blacklistedSessions.putIfAbsent(externalId, sessionRef) == null) {
@ -315,6 +314,8 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr
log.debug("[{}][{}][{}] Session is no longer blacklisted.", sessionRef.getSecurityCtx().getTenantId(), sessionRef.getSecurityCtx().getId(), externalId);
blacklistedSessions.remove(externalId);
}
} else {
perSessionUpdateLimits.remove(sessionRef.getSessionId());
}
sessionMd.sendMsg(msg);
} else {
@ -360,8 +361,7 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr
private boolean checkLimits(WebSocketSession session, TelemetryWebSocketSessionRef sessionRef) throws Exception {
var tenantProfileConfiguration =
tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultTenantProfileConfiguration();
tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration();
if (tenantProfileConfiguration == null) {
return true;
}
@ -428,7 +428,7 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr
}
private void cleanupLimits(WebSocketSession session, TelemetryWebSocketSessionRef sessionRef) {
var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultTenantProfileConfiguration();
var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration();
String sessionId = session.getId();
perSessionUpdateLimits.remove(sessionRef.getSessionId());

34
application/src/main/java/org/thingsboard/server/service/install/update/RateLimitsUpdater.java
View File

@ -28,39 +28,39 @@ import org.thingsboard.server.dao.tenant.TenantProfileService;
@Component
class RateLimitsUpdater extends PaginatedUpdater<String, TenantProfile> {
@Value("${server.rest.limits.tenant.enabled}")
@Value("#{ environment.getProperty('TB_SERVER_REST_LIMITS_TENANT_ENABLED') ?: environment.getProperty('server.rest.limits.tenant.enabled') ?: 'false' }")
boolean tenantServerRestLimitsEnabled;
@Value("${server.rest.limits.tenant.configuration}")
@Value("#{ environment.getProperty('TB_SERVER_REST_LIMITS_TENANT_CONFIGURATION') ?: environment.getProperty('server.rest.limits.tenant.configuration') ?: '100:1,2000:60' }")
String tenantServerRestLimitsConfiguration;
@Value("${server.rest.limits.customer.enabled}")
@Value("#{ environment.getProperty('TB_SERVER_REST_LIMITS_CUSTOMER_ENABLED') ?: environment.getProperty('server.rest.limits.customer.enabled') ?: 'false' }")
boolean customerServerRestLimitsEnabled;
@Value("${server.rest.limits.customer.configuration}")
@Value("#{ environment.getProperty('TB_SERVER_REST_LIMITS_CUSTOMER_CONFIGURATION') ?: environment.getProperty('server.rest.limits.customer.configuration') ?: '50:1,1000:60' }")
String customerServerRestLimitsConfiguration;
@Value("${server.ws.limits.max_sessions_per_tenant}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_TENANT') ?: environment.getProperty('server.ws.limits.max_sessions_per_tenant') ?: '0' }")
private int maxWsSessionsPerTenant;
@Value("${server.ws.limits.max_sessions_per_customer}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_CUSTOMER') ?: environment.getProperty('server.ws.limits.max_sessions_per_customer') ?: '0' }")
private int maxWsSessionsPerCustomer;
@Value("${server.ws.limits.max_sessions_per_regular_user}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_REGULAR_USER') ?: environment.getProperty('server.ws.limits.max_sessions_per_regular_user') ?: '0' }")
private int maxWsSessionsPerRegularUser;
@Value("${server.ws.limits.max_sessions_per_public_user}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_PUBLIC_USER') ?: environment.getProperty('server.ws.limits.max_sessions_per_public_user') ?: '0' }")
private int maxWsSessionsPerPublicUser;
@Value("${server.ws.limits.max_queue_per_ws_session}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_QUEUE_PER_WS_SESSION') ?: environment.getProperty('server.ws.limits.max_queue_per_ws_session') ?: '500' }")
private int wsMsgQueueLimitPerSession;
@Value("${server.ws.limits.max_subscriptions_per_tenant}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_TENANT') ?: environment.getProperty('server.ws.limits.max_subscriptions_per_tenant') ?: '0' }")
private long maxWsSubscriptionsPerTenant;
@Value("${server.ws.limits.max_subscriptions_per_customer}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_CUSTOMER') ?: environment.getProperty('server.ws.limits.max_subscriptions_per_customer') ?: '0' }")
private long maxWsSubscriptionsPerCustomer;
@Value("${server.ws.limits.max_subscriptions_per_regular_user}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_REGULAR_USER') ?: environment.getProperty('server.ws.limits.max_subscriptions_per_regular_user') ?: '0' }")
private long maxWsSubscriptionsPerRegularUser;
@Value("${server.ws.limits.max_subscriptions_per_public_user}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_PUBLIC_USER') ?: environment.getProperty('server.ws.limits.max_subscriptions_per_public_user') ?: '0' }")
private long maxWsSubscriptionsPerPublicUser;
@Value("${server.ws.limits.max_updates_per_session}")
@Value("#{ environment.getProperty('TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_UPDATES_PER_SESSION') ?: environment.getProperty('server.ws.limits.max_updates_per_session') ?: '300:1,3000:60' }")
private String wsUpdatesPerSessionRateLimit;
@Value("${cassandra.query.tenant_rate_limits.enabled}")
@Value("#{ environment.getProperty('CASSANDRA_QUERY_TENANT_RATE_LIMITS_ENABLED') ?: environment.getProperty('cassandra.query.tenant_rate_limits.enabled') ?: 'false' }")
private boolean cassandraQueryTenantRateLimitsEnabled;
@Value("${cassandra.query.tenant_rate_limits.configuration}")
@Value("#{ environment.getProperty('CASSANDRA_QUERY_TENANT_RATE_LIMITS_CONFIGURATION') ?: environment.getProperty('cassandra.query.tenant_rate_limits.configuration') ?: '1000:1,30000:60' }")
private String cassandraQueryTenantRateLimitsConfiguration;
@Autowired
@ -83,7 +83,7 @@ class RateLimitsUpdater extends PaginatedUpdater<String, TenantProfile> {
@Override
protected void updateEntity(TenantProfile tenantProfile) {
var profileConfiguration = tenantProfile.getDefaultTenantProfileConfiguration();
var profileConfiguration = tenantProfile.getDefaultProfileConfiguration();
if (tenantServerRestLimitsEnabled && StringUtils.isNotEmpty(tenantServerRestLimitsConfiguration)) {
profileConfiguration.setTenantServerRestLimitsConfiguration(tenantServerRestLimitsConfiguration);

4
application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java
View File

@ -304,7 +304,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
}
private void processSessionClose(TelemetryWebSocketSessionRef sessionRef) {
var tenantProfileConfiguration = (DefaultTenantProfileConfiguration) tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultTenantProfileConfiguration();
var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration();
if (tenantProfileConfiguration != null) {
String sessionId = "[" + sessionRef.getSessionId() + "]";
@ -338,7 +338,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
}
private boolean processSubscription(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd) {
var tenantProfileConfiguration = (DefaultTenantProfileConfiguration) tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultTenantProfileConfiguration();
var tenantProfileConfiguration = (DefaultTenantProfileConfiguration) tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration();
String subId = "[" + sessionRef.getSessionId() + "]:[" + cmd.getCmdId() + "]";
try {

21
application/src/main/resources/thingsboard.yml
View File

@ -56,18 +56,6 @@ server:
send_timeout: "${TB_SERVER_WS_SEND_TIMEOUT:5000}"
# recommended timeout >= 30 seconds. Platform will attempt to send 'ping' request 3 times within the timeout
ping_timeout: "${TB_SERVER_WS_PING_TIMEOUT:30000}"
limits:
# Limit the amount of sessions and subscriptions available on each server. Put values to zero to disable particular limitation
max_sessions_per_tenant: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_TENANT:0}"
max_sessions_per_customer: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_CUSTOMER:0}"
max_sessions_per_regular_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_REGULAR_USER:0}"
max_sessions_per_public_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_PUBLIC_USER:0}"
max_queue_per_ws_session: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_QUEUE_PER_WS_SESSION:500}"
max_subscriptions_per_tenant: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_TENANT:0}"
max_subscriptions_per_customer: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_CUSTOMER:0}"
max_subscriptions_per_regular_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_REGULAR_USER:0}"
max_subscriptions_per_public_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_PUBLIC_USER:0}"
max_updates_per_session: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_UPDATES_PER_SESSION:300:1,3000:60}"
dynamic_page_link:
refresh_interval: "${TB_SERVER_WS_DYNAMIC_PAGE_LINK_REFRESH_INTERVAL_SEC:60}"
refresh_pool_size: "${TB_SERVER_WS_DYNAMIC_PAGE_LINK_REFRESH_POOL_SIZE:1}"
@ -76,13 +64,6 @@ server:
max_entities_per_data_subscription: "${TB_SERVER_WS_MAX_ENTITIES_PER_DATA_SUBSCRIPTION:10000}"
max_entities_per_alarm_subscription: "${TB_SERVER_WS_MAX_ENTITIES_PER_ALARM_SUBSCRIPTION:10000}"
rest:
limits:
tenant:
enabled: "${TB_SERVER_REST_LIMITS_TENANT_ENABLED:false}"
configuration: "${TB_SERVER_REST_LIMITS_TENANT_CONFIGURATION:100:1,2000:60}"
customer:
enabled: "${TB_SERVER_REST_LIMITS_CUSTOMER_ENABLED:false}"
configuration: "${TB_SERVER_REST_LIMITS_CUSTOMER_CONFIGURATION:50:1,1000:60}"
server_side_rpc:
# Minimum value of the server side RPC timeout. May override value provided in the REST API call.
# Since 2.5 migration to queues, the RPC delay depends on the size of the pending messages in the queue,
@ -255,8 +236,6 @@ cassandra:
# log one of cassandra queries with specified frequency (0 - logging is disabled)
print_queries_freq: "${CASSANDRA_QUERY_PRINT_FREQ:0}"
tenant_rate_limits:
enabled: "${CASSANDRA_QUERY_TENANT_RATE_LIMITS_ENABLED:false}"
configuration: "${CASSANDRA_QUERY_TENANT_RATE_LIMITS_CONFIGURATION:1000:1,30000:60}"
print_tenant_names: "${CASSANDRA_QUERY_TENANT_RATE_LIMITS_PRINT_TENANT_NAMES:false}"
# SQL configuration parameters

10
common/data/src/main/java/org/thingsboard/server/common/data/TenantProfile.java
View File

@ -130,6 +130,11 @@ public class TenantProfile extends SearchTextBased<TenantProfileId> implements H
.map(profileConfiguration -> (DefaultTenantProfileConfiguration) profileConfiguration);
}
@JsonIgnore
public DefaultTenantProfileConfiguration getDefaultProfileConfiguration() {
return getProfileConfiguration().orElse(null);
}
public TenantProfileData createDefaultTenantProfileData() {
TenantProfileData tpd = new TenantProfileData();
tpd.setConfiguration(new DefaultTenantProfileConfiguration());
@ -146,9 +151,4 @@ public class TenantProfile extends SearchTextBased<TenantProfileId> implements H
}
}
@JsonIgnore
public DefaultTenantProfileConfiguration getDefaultTenantProfileConfiguration() {
return getProfileConfiguration().orElse(null);
}
}

20
common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java
View File

@ -46,6 +46,16 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura
private String transportDeviceTelemetryMsgRateLimit;
private String transportDeviceTelemetryDataPointsRateLimit;
private long maxTransportMessages;
private long maxTransportDataPoints;
private long maxREExecutions;
private long maxJSExecutions;
private long maxDPStorageDays;
private int maxRuleNodeExecutionsPerMessage;
private long maxEmails;
private long maxSms;
private long maxCreatedAlarms;
private String tenantServerRestLimitsConfiguration;
private String customerServerRestLimitsConfiguration;
@ -62,16 +72,6 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura
private String cassandraQueryTenantRateLimitsConfiguration;
private long maxTransportMessages;
private long maxTransportDataPoints;
private long maxREExecutions;
private long maxJSExecutions;
private long maxDPStorageDays;
private int maxRuleNodeExecutionsPerMessage;
private long maxEmails;
private long maxSms;
private long maxCreatedAlarms;
private int defaultStorageTtlDays;
private int alarmsTtlDays;
private int rpcTtlDays;

4
dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java
View File

@ -116,7 +116,7 @@ public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extend
F result = wrap(task, settableFuture);
boolean perTenantLimitReached = false;
var tenantProfileConfiguration = tenantProfileCache.get(task.getTenantId()).getDefaultTenantProfileConfiguration();
var tenantProfileConfiguration = tenantProfileCache.get(task.getTenantId()).getDefaultProfileConfiguration();
if (StringUtils.isNotEmpty(tenantProfileConfiguration.getCassandraQueryTenantRateLimitsConfiguration())) {
if (task.getTenantId() == null) {
log.info("Invalid task received: {}", task);
@ -131,6 +131,8 @@ public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extend
perTenantLimitReached = true;
}
}
} else if (!TenantId.SYS_TENANT_ID.equals(task.getTenantId())) {
perTenantLimits.remove(task.getTenantId());
}
if (!perTenantLimitReached) {

Write Preview
Loading…
Cancel
Save