|
|
@ -5,7 +5,7 @@ |
|
|
* you may not use this file except in compliance with the License. |
|
|
* you may not use this file except in compliance with the License. |
|
|
* You may obtain a copy of the License at |
|
|
* You may obtain a copy of the License at |
|
|
* |
|
|
* |
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
|
* |
|
|
* |
|
|
* Unless required by applicable law or agreed to in writing, software |
|
|
* Unless required by applicable law or agreed to in writing, software |
|
|
* distributed under the License is distributed on an "AS IS" BASIS, |
|
|
* distributed under the License is distributed on an "AS IS" BASIS, |
|
|
@ -17,76 +17,96 @@ package org.thingsboard.server.common.transport.limits; |
|
|
|
|
|
|
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import org.springframework.stereotype.Service; |
|
|
import org.springframework.stereotype.Service; |
|
|
|
|
|
import org.springframework.util.StringUtils; |
|
|
|
|
|
import org.thingsboard.server.common.data.EntityType; |
|
|
import org.thingsboard.server.common.data.TenantProfile; |
|
|
import org.thingsboard.server.common.data.TenantProfile; |
|
|
import org.thingsboard.server.common.data.id.DeviceId; |
|
|
import org.thingsboard.server.common.data.id.DeviceId; |
|
|
import org.thingsboard.server.common.data.id.TenantId; |
|
|
import org.thingsboard.server.common.data.id.TenantId; |
|
|
|
|
|
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.TenantProfileData; |
|
|
|
|
|
import org.thingsboard.server.common.msg.tools.TbRateLimits; |
|
|
import org.thingsboard.server.common.transport.TransportTenantProfileCache; |
|
|
import org.thingsboard.server.common.transport.TransportTenantProfileCache; |
|
|
import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult; |
|
|
import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult; |
|
|
import org.thingsboard.server.queue.util.TbTransportComponent; |
|
|
import org.thingsboard.server.queue.util.TbTransportComponent; |
|
|
|
|
|
|
|
|
|
|
|
import java.util.HashSet; |
|
|
|
|
|
import java.util.Set; |
|
|
import java.util.concurrent.ConcurrentHashMap; |
|
|
import java.util.concurrent.ConcurrentHashMap; |
|
|
import java.util.concurrent.ConcurrentMap; |
|
|
import java.util.concurrent.ConcurrentMap; |
|
|
|
|
|
import java.util.function.BiConsumer; |
|
|
|
|
|
import java.util.function.Function; |
|
|
|
|
|
|
|
|
@Service |
|
|
@Service |
|
|
@TbTransportComponent |
|
|
@TbTransportComponent |
|
|
@Slf4j |
|
|
@Slf4j |
|
|
public class DefaultTransportRateLimitService implements TransportRateLimitService { |
|
|
public class DefaultTransportRateLimitService implements TransportRateLimitService { |
|
|
|
|
|
|
|
|
|
|
|
private final static DummyTransportRateLimit ALLOW = new DummyTransportRateLimit(); |
|
|
private final ConcurrentMap<TenantId, Boolean> tenantAllowed = new ConcurrentHashMap<>(); |
|
|
private final ConcurrentMap<TenantId, Boolean> tenantAllowed = new ConcurrentHashMap<>(); |
|
|
private final ConcurrentMap<TenantId, TransportRateLimit[]> perTenantLimits = new ConcurrentHashMap<>(); |
|
|
private final ConcurrentMap<TenantId, Set<DeviceId>> tenantDevices = new ConcurrentHashMap<>(); |
|
|
private final ConcurrentMap<DeviceId, TransportRateLimit[]> perDeviceLimits = new ConcurrentHashMap<>(); |
|
|
private final ConcurrentMap<TenantId, EntityTransportRateLimits> perTenantLimits = new ConcurrentHashMap<>(); |
|
|
|
|
|
private final ConcurrentMap<DeviceId, EntityTransportRateLimits> perDeviceLimits = new ConcurrentHashMap<>(); |
|
|
|
|
|
|
|
|
private final TransportRateLimitFactory rateLimitFactory; |
|
|
|
|
|
private final TransportTenantProfileCache tenantProfileCache; |
|
|
private final TransportTenantProfileCache tenantProfileCache; |
|
|
|
|
|
|
|
|
public DefaultTransportRateLimitService(TransportRateLimitFactory rateLimitFactory, TransportTenantProfileCache tenantProfileCache) { |
|
|
public DefaultTransportRateLimitService(TransportTenantProfileCache tenantProfileCache) { |
|
|
this.rateLimitFactory = rateLimitFactory; |
|
|
|
|
|
this.tenantProfileCache = tenantProfileCache; |
|
|
this.tenantProfileCache = tenantProfileCache; |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
public TransportRateLimitType checkLimits(TenantId tenantId, DeviceId deviceId, int dataPoints, TransportRateLimitType... limits) { |
|
|
public EntityType checkLimits(TenantId tenantId, DeviceId deviceId, int dataPoints) { |
|
|
if (!tenantAllowed.getOrDefault(tenantId, Boolean.TRUE)) { |
|
|
if (!tenantAllowed.getOrDefault(tenantId, Boolean.TRUE)) { |
|
|
return TransportRateLimitType.TENANT_ADDED_TO_DISABLED_LIST; |
|
|
return EntityType.TENANT; |
|
|
} |
|
|
} |
|
|
TransportRateLimit[] tenantLimits = getTenantRateLimits(tenantId); |
|
|
if (!checkEntityRateLimit(dataPoints, getTenantRateLimits(tenantId))) { |
|
|
TransportRateLimit[] deviceLimits = getDeviceRateLimits(tenantId, deviceId); |
|
|
return EntityType.TENANT; |
|
|
for (TransportRateLimitType limitType : limits) { |
|
|
} |
|
|
TransportRateLimit rateLimit; |
|
|
if (!checkEntityRateLimit(dataPoints, getDeviceRateLimits(tenantId, deviceId))) { |
|
|
if (limitType.isTenantLevel()) { |
|
|
return EntityType.DEVICE; |
|
|
rateLimit = tenantLimits[limitType.ordinal()]; |
|
|
|
|
|
} else { |
|
|
|
|
|
rateLimit = deviceLimits[limitType.ordinal()]; |
|
|
|
|
|
} |
|
|
|
|
|
if (!rateLimit.tryConsume(limitType.isMessageLevel() ? 1L : dataPoints)) { |
|
|
|
|
|
return limitType; |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
} |
|
|
return null; |
|
|
return null; |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private boolean checkEntityRateLimit(int dataPoints, EntityTransportRateLimits tenantLimits) { |
|
|
|
|
|
if (dataPoints > 0) { |
|
|
|
|
|
return tenantLimits.getTelemetryMsgRateLimit().tryConsume() && tenantLimits.getTelemetryDataPointsRateLimit().tryConsume(dataPoints); |
|
|
|
|
|
} else { |
|
|
|
|
|
return tenantLimits.getRegularMsgRateLimit().tryConsume(); |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
public void update(TenantProfileUpdateResult update) { |
|
|
public void update(TenantProfileUpdateResult update) { |
|
|
TransportRateLimit[] newLimits = createTransportRateLimits(update.getProfile()); |
|
|
EntityTransportRateLimits tenantRateLimitPrototype = createRateLimits(update.getProfile(), true); |
|
|
|
|
|
EntityTransportRateLimits deviceRateLimitPrototype = createRateLimits(update.getProfile(), false); |
|
|
for (TenantId tenantId : update.getAffectedTenants()) { |
|
|
for (TenantId tenantId : update.getAffectedTenants()) { |
|
|
mergeLimits(tenantId, newLimits); |
|
|
mergeLimits(tenantId, tenantRateLimitPrototype, perTenantLimits::get, perTenantLimits::put); |
|
|
|
|
|
tenantDevices.get(tenantId).forEach(deviceId -> { |
|
|
|
|
|
mergeLimits(deviceId, deviceRateLimitPrototype, perDeviceLimits::get, perDeviceLimits::put); |
|
|
|
|
|
}); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
public void update(TenantId tenantId) { |
|
|
public void update(TenantId tenantId) { |
|
|
mergeLimits(tenantId, fetchProfileAndInit(tenantId)); |
|
|
EntityTransportRateLimits tenantRateLimitPrototype = createRateLimits(tenantProfileCache.get(tenantId), true); |
|
|
|
|
|
EntityTransportRateLimits deviceRateLimitPrototype = createRateLimits(tenantProfileCache.get(tenantId), false); |
|
|
|
|
|
mergeLimits(tenantId, tenantRateLimitPrototype, perTenantLimits::get, perTenantLimits::put); |
|
|
|
|
|
tenantDevices.get(tenantId).forEach(deviceId -> { |
|
|
|
|
|
mergeLimits(deviceId, deviceRateLimitPrototype, perDeviceLimits::get, perDeviceLimits::put); |
|
|
|
|
|
}); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
public void remove(TenantId tenantId) { |
|
|
public void remove(TenantId tenantId) { |
|
|
perTenantLimits.remove(tenantId); |
|
|
perTenantLimits.remove(tenantId); |
|
|
|
|
|
tenantDevices.remove(tenantId); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
public void remove(DeviceId deviceId) { |
|
|
public void remove(DeviceId deviceId) { |
|
|
perDeviceLimits.remove(deviceId); |
|
|
perDeviceLimits.remove(deviceId); |
|
|
|
|
|
tenantDevices.values().forEach(set -> set.remove(deviceId)); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
@ -94,48 +114,66 @@ public class DefaultTransportRateLimitService implements TransportRateLimitServi |
|
|
tenantAllowed.put(tenantId, allowed); |
|
|
tenantAllowed.put(tenantId, allowed); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private void mergeLimits(TenantId tenantId, TransportRateLimit[] newRateLimits) { |
|
|
private <T> void mergeLimits(T deviceId, EntityTransportRateLimits newRateLimits, |
|
|
TransportRateLimit[] oldRateLimits = perTenantLimits.get(tenantId); |
|
|
Function<T, EntityTransportRateLimits> getFunction, |
|
|
|
|
|
BiConsumer<T, EntityTransportRateLimits> putFunction) { |
|
|
|
|
|
EntityTransportRateLimits oldRateLimits = getFunction.apply(deviceId); |
|
|
if (oldRateLimits == null) { |
|
|
if (oldRateLimits == null) { |
|
|
perTenantLimits.put(tenantId, newRateLimits); |
|
|
putFunction.accept(deviceId, newRateLimits); |
|
|
} else { |
|
|
} else { |
|
|
for (int i = 0; i < TransportRateLimitType.values().length; i++) { |
|
|
EntityTransportRateLimits updated = merge(oldRateLimits, newRateLimits); |
|
|
TransportRateLimit newLimit = newRateLimits[i]; |
|
|
if (updated != null) { |
|
|
TransportRateLimit oldLimit = oldRateLimits[i]; |
|
|
putFunction.accept(deviceId, updated); |
|
|
if (newLimit != null && (oldLimit == null || !oldLimit.getConfiguration().equals(newLimit.getConfiguration()))) { |
|
|
|
|
|
oldRateLimits[i] = newLimit; |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private TransportRateLimit[] fetchProfileAndInit(TenantId tenantId) { |
|
|
private EntityTransportRateLimits merge(EntityTransportRateLimits oldRateLimits, EntityTransportRateLimits newRateLimits) { |
|
|
return perTenantLimits.computeIfAbsent(tenantId, tmp -> createTransportRateLimits(tenantProfileCache.get(tenantId))); |
|
|
boolean regularUpdate = !oldRateLimits.getRegularMsgRateLimit().getConfiguration().equals(newRateLimits.getRegularMsgRateLimit().getConfiguration()); |
|
|
|
|
|
boolean telemetryMsgRateUpdate = !oldRateLimits.getTelemetryMsgRateLimit().getConfiguration().equals(newRateLimits.getTelemetryMsgRateLimit().getConfiguration()); |
|
|
|
|
|
boolean telemetryDataPointUpdate = !oldRateLimits.getTelemetryDataPointsRateLimit().getConfiguration().equals(newRateLimits.getTelemetryDataPointsRateLimit().getConfiguration()); |
|
|
|
|
|
if (regularUpdate || telemetryMsgRateUpdate || telemetryDataPointUpdate) { |
|
|
|
|
|
return new EntityTransportRateLimits( |
|
|
|
|
|
regularUpdate ? newLimit(newRateLimits.getRegularMsgRateLimit().getConfiguration()) : oldRateLimits.getRegularMsgRateLimit(), |
|
|
|
|
|
telemetryMsgRateUpdate ? newLimit(newRateLimits.getTelemetryMsgRateLimit().getConfiguration()) : oldRateLimits.getTelemetryMsgRateLimit(), |
|
|
|
|
|
telemetryDataPointUpdate ? newLimit(newRateLimits.getTelemetryDataPointsRateLimit().getConfiguration()) : oldRateLimits.getTelemetryDataPointsRateLimit()); |
|
|
|
|
|
} else { |
|
|
|
|
|
return null; |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private TransportRateLimit[] createTransportRateLimits(TenantProfile tenantProfile) { |
|
|
private EntityTransportRateLimits createRateLimits(TenantProfile tenantProfile, boolean tenant) { |
|
|
TenantProfileData profileData = tenantProfile.getProfileData(); |
|
|
TenantProfileData profileData = tenantProfile.getProfileData(); |
|
|
TransportRateLimit[] rateLimits = new TransportRateLimit[TransportRateLimitType.values().length]; |
|
|
DefaultTenantProfileConfiguration profile = (DefaultTenantProfileConfiguration) profileData.getConfiguration(); |
|
|
for (TransportRateLimitType type : TransportRateLimitType.values()) { |
|
|
if (profile == null) { |
|
|
rateLimits[type.ordinal()] = rateLimitFactory.create(type, profileData.getProperties().get(type.getConfigurationKey())); |
|
|
return new EntityTransportRateLimits(ALLOW, ALLOW, ALLOW); |
|
|
|
|
|
} else { |
|
|
|
|
|
TransportRateLimit regularMsgRateLimit = newLimit(tenant ? profile.getTransportTenantMsgRateLimit() : profile.getTransportDeviceMsgRateLimit()); |
|
|
|
|
|
TransportRateLimit telemetryMsgRateLimit = newLimit(tenant ? profile.getTransportTenantTelemetryMsgRateLimit() : profile.getTransportDeviceTelemetryMsgRateLimit()); |
|
|
|
|
|
TransportRateLimit telemetryDpRateLimit = newLimit(tenant ? profile.getTransportTenantTelemetryDataPointsRateLimit() : profile.getTransportTenantTelemetryDataPointsRateLimit()); |
|
|
|
|
|
return new EntityTransportRateLimits(regularMsgRateLimit, telemetryMsgRateLimit, telemetryDpRateLimit); |
|
|
} |
|
|
} |
|
|
return rateLimits; |
|
|
|
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private TransportRateLimit[] getTenantRateLimits(TenantId tenantId) { |
|
|
private static TransportRateLimit newLimit(String config) { |
|
|
TransportRateLimit[] limits = perTenantLimits.get(tenantId); |
|
|
return StringUtils.isEmpty(config) ? ALLOW : new SimpleTransportRateLimit(config); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private EntityTransportRateLimits getTenantRateLimits(TenantId tenantId) { |
|
|
|
|
|
EntityTransportRateLimits limits = perTenantLimits.get(tenantId); |
|
|
if (limits == null) { |
|
|
if (limits == null) { |
|
|
limits = fetchProfileAndInit(tenantId); |
|
|
limits = createRateLimits(tenantProfileCache.get(tenantId), true); |
|
|
perTenantLimits.put(tenantId, limits); |
|
|
perTenantLimits.put(tenantId, limits); |
|
|
} |
|
|
} |
|
|
return limits; |
|
|
return limits; |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private TransportRateLimit[] getDeviceRateLimits(TenantId tenantId, DeviceId deviceId) { |
|
|
private EntityTransportRateLimits getDeviceRateLimits(TenantId tenantId, DeviceId deviceId) { |
|
|
TransportRateLimit[] limits = perDeviceLimits.get(deviceId); |
|
|
EntityTransportRateLimits limits = perDeviceLimits.get(deviceId); |
|
|
if (limits == null) { |
|
|
if (limits == null) { |
|
|
limits = fetchProfileAndInit(tenantId); |
|
|
limits = createRateLimits(tenantProfileCache.get(tenantId), false); |
|
|
perDeviceLimits.put(deviceId, limits); |
|
|
perDeviceLimits.put(deviceId, limits); |
|
|
|
|
|
tenantDevices.computeIfAbsent(tenantId, id -> ConcurrentHashMap.newKeySet()).add(deviceId); |
|
|
} |
|
|
} |
|
|
return limits; |
|
|
return limits; |
|
|
} |
|
|
} |
|
|
|
Andrii Shvaika
6 years ago