Browse Source
Fixed transport tenant-profile lock convoy under cold-cache reconnect stormpull/15767/head
Viacheslav Klimov
4 months ago
committed by
GitHub
GitHub
12 changed files with 538 additions and 97 deletions
@ -0,0 +1,202 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.transport.limits; |
||||
|
|
||||
|
import org.junit.jupiter.api.AfterEach; |
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.junit.jupiter.params.ParameterizedTest; |
||||
|
import org.junit.jupiter.params.provider.EnumSource; |
||||
|
import org.thingsboard.server.common.data.TenantProfile; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.id.TenantProfileId; |
||||
|
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; |
||||
|
import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; |
||||
|
import org.thingsboard.server.common.transport.TransportTenantProfileCache; |
||||
|
import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult; |
||||
|
|
||||
|
import java.util.Set; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.CountDownLatch; |
||||
|
import java.util.concurrent.ExecutorService; |
||||
|
import java.util.concurrent.Executors; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
|
||||
|
class DefaultTransportRateLimitServiceTest { |
||||
|
|
||||
|
private TransportTenantProfileCache tenantProfileCache; |
||||
|
private ExecutorService executor; |
||||
|
|
||||
|
private final TenantId tenant = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
|
||||
|
@BeforeEach |
||||
|
void setUp() { |
||||
|
tenantProfileCache = mock(TransportTenantProfileCache.class); |
||||
|
executor = Executors.newCachedThreadPool(); |
||||
|
} |
||||
|
|
||||
|
@AfterEach |
||||
|
void tearDown() { |
||||
|
executor.shutdownNow(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void checkLimitsDoesNotHoldMapBinLockAcrossProfileFetch() throws Exception { |
||||
|
// Two concurrent rate-limit checks for the SAME tenant must both be able to reach
|
||||
|
// the (blocking) tenant-profile fetch concurrently. If the blocking fetch runs inside
|
||||
|
// ConcurrentHashMap.computeIfAbsent, the second caller is stuck on the bin reservation
|
||||
|
// node and never reaches the fetch -> the latch never reaches zero.
|
||||
|
CountDownLatch bothCallersReachedFetch = new CountDownLatch(2); |
||||
|
CountDownLatch releaseFetch = new CountDownLatch(1); |
||||
|
|
||||
|
when(tenantProfileCache.get(tenant)).thenAnswer(invocation -> { |
||||
|
bothCallersReachedFetch.countDown(); |
||||
|
releaseFetch.await(5, TimeUnit.SECONDS); |
||||
|
return tenantProfile(); |
||||
|
}); |
||||
|
|
||||
|
DefaultTransportRateLimitService service = new DefaultTransportRateLimitService(tenantProfileCache); |
||||
|
|
||||
|
Runnable check = () -> service.checkLimits(tenant, null, null, 1, false); |
||||
|
executor.submit(check); |
||||
|
executor.submit(check); |
||||
|
|
||||
|
boolean bothReached = bothCallersReachedFetch.await(3, TimeUnit.SECONDS); |
||||
|
releaseFetch.countDown(); |
||||
|
|
||||
|
assertThat(bothReached) |
||||
|
.as("both checkLimits calls should reach the profile fetch concurrently (no bin lock across I/O)") |
||||
|
.isTrue(); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@EnumSource(TransportLimitsType.class) |
||||
|
void eachLimitsTypeReadsItsOwnProfileFields(TransportLimitsType type) { |
||||
|
// Distinct sentinel per profile field so a transposed method reference (e.g. GATEWAY_DEVICE_LIMITS
|
||||
|
// wired to the plain gateway getters) resolves to the wrong value and fails the assertion.
|
||||
|
DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); |
||||
|
config.setTransportTenantMsgRateLimit("tenant-msg"); |
||||
|
config.setTransportTenantTelemetryMsgRateLimit("tenant-tele-msg"); |
||||
|
config.setTransportTenantTelemetryDataPointsRateLimit("tenant-tele-dp"); |
||||
|
config.setTransportDeviceMsgRateLimit("device-msg"); |
||||
|
config.setTransportDeviceTelemetryMsgRateLimit("device-tele-msg"); |
||||
|
config.setTransportDeviceTelemetryDataPointsRateLimit("device-tele-dp"); |
||||
|
config.setTransportGatewayMsgRateLimit("gateway-msg"); |
||||
|
config.setTransportGatewayTelemetryMsgRateLimit("gateway-tele-msg"); |
||||
|
config.setTransportGatewayTelemetryDataPointsRateLimit("gateway-tele-dp"); |
||||
|
config.setTransportGatewayDeviceMsgRateLimit("gateway-device-msg"); |
||||
|
config.setTransportGatewayDeviceTelemetryMsgRateLimit("gateway-device-tele-msg"); |
||||
|
config.setTransportGatewayDeviceTelemetryDataPointsRateLimit("gateway-device-tele-dp"); |
||||
|
|
||||
|
String prefix = switch (type) { |
||||
|
case TENANT_LIMITS -> "tenant"; |
||||
|
case DEVICE_LIMITS -> "device"; |
||||
|
case GATEWAY_LIMITS -> "gateway"; |
||||
|
case GATEWAY_DEVICE_LIMITS -> "gateway-device"; |
||||
|
}; |
||||
|
|
||||
|
assertThat(type.getRegularMsgRateLimit().apply(config)).isEqualTo(prefix + "-msg"); |
||||
|
assertThat(type.getTelemetryMsgRateLimit().apply(config)).isEqualTo(prefix + "-tele-msg"); |
||||
|
assertThat(type.getTelemetryDataPointsRateLimit().apply(config)).isEqualTo(prefix + "-tele-dp"); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@EnumSource(EntityLevel.class) |
||||
|
void profileUpdateReachesEntityTrackedDuringFirstCheck(EntityLevel level) { |
||||
|
DeviceId entity = new DeviceId(UUID.randomUUID()); |
||||
|
when(tenantProfileCache.get(tenant)).thenReturn(profileWithRegularMsgLimit(level, "100:600")); |
||||
|
DefaultTransportRateLimitService service = new DefaultTransportRateLimitService(tenantProfileCache); |
||||
|
|
||||
|
// First check resolves the (permissive) limit and must register the entity into the per-tenant
|
||||
|
// tracking set via the onMiss callback - otherwise a later update(tenantId) can't reach it.
|
||||
|
assertThat(level.check(service, tenant, entity)) |
||||
|
.as("permissive limit should allow the first %s check", level).isNull(); |
||||
|
|
||||
|
// Tighten the limit to a single message and push a profile update for this tenant.
|
||||
|
service.update(new TenantProfileUpdateResult(profileWithRegularMsgLimit(level, "1:600"), Set.of(tenant))); |
||||
|
|
||||
|
// The freshly merged "1:600" bucket allows exactly one message...
|
||||
|
assertThat(level.check(service, tenant, entity)).isNull(); |
||||
|
// ...and blocks the next one. This only happens if update(tenantId) reached the tracked entity.
|
||||
|
assertThat(level.check(service, tenant, entity)) |
||||
|
.as("update(tenantId) must reach the tracked %s so the tightened limit applies", level).isNotNull(); |
||||
|
} |
||||
|
|
||||
|
private TenantProfile tenantProfile() { |
||||
|
return profileWith(new DefaultTenantProfileConfiguration()); |
||||
|
} |
||||
|
|
||||
|
private TenantProfile profileWithRegularMsgLimit(EntityLevel level, String regularMsgRateLimit) { |
||||
|
DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); |
||||
|
level.setRegularMsgRateLimit(config, regularMsgRateLimit); |
||||
|
return profileWith(config); |
||||
|
} |
||||
|
|
||||
|
private TenantProfile profileWith(DefaultTenantProfileConfiguration config) { |
||||
|
TenantProfile profile = new TenantProfile(new TenantProfileId(UUID.randomUUID())); |
||||
|
profile.setName("test-profile"); |
||||
|
TenantProfileData profileData = new TenantProfileData(); |
||||
|
profileData.setConfiguration(config); |
||||
|
profile.setProfileData(profileData); |
||||
|
return profile; |
||||
|
} |
||||
|
|
||||
|
private enum EntityLevel { |
||||
|
DEVICE { |
||||
|
@Override |
||||
|
void setRegularMsgRateLimit(DefaultTenantProfileConfiguration config, String value) { |
||||
|
config.setTransportDeviceMsgRateLimit(value); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
Object check(DefaultTransportRateLimitService service, TenantId tenantId, DeviceId entityId) { |
||||
|
return service.checkLimits(tenantId, null, entityId, 0, false); |
||||
|
} |
||||
|
}, |
||||
|
GATEWAY { |
||||
|
@Override |
||||
|
void setRegularMsgRateLimit(DefaultTenantProfileConfiguration config, String value) { |
||||
|
config.setTransportGatewayMsgRateLimit(value); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
Object check(DefaultTransportRateLimitService service, TenantId tenantId, DeviceId entityId) { |
||||
|
return service.checkLimits(tenantId, entityId, null, 0, false); |
||||
|
} |
||||
|
}, |
||||
|
GATEWAY_DEVICE { |
||||
|
@Override |
||||
|
void setRegularMsgRateLimit(DefaultTenantProfileConfiguration config, String value) { |
||||
|
config.setTransportGatewayDeviceMsgRateLimit(value); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
Object check(DefaultTransportRateLimitService service, TenantId tenantId, DeviceId entityId) { |
||||
|
return service.checkLimits(tenantId, null, entityId, 0, true); |
||||
|
} |
||||
|
}; |
||||
|
|
||||
|
abstract void setRegularMsgRateLimit(DefaultTenantProfileConfiguration config, String value); |
||||
|
|
||||
|
abstract Object check(DefaultTransportRateLimitService service, TenantId tenantId, DeviceId entityId); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,191 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.transport.service; |
||||
|
|
||||
|
import com.google.common.util.concurrent.Striped; |
||||
|
import org.junit.jupiter.api.AfterEach; |
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.thingsboard.server.common.data.ApiUsageState; |
||||
|
import org.thingsboard.server.common.data.ApiUsageStateValue; |
||||
|
import org.thingsboard.server.common.data.EntityType; |
||||
|
import org.thingsboard.server.common.data.TenantProfile; |
||||
|
import org.thingsboard.server.common.data.id.ApiUsageStateId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.id.TenantProfileId; |
||||
|
import org.thingsboard.server.common.transport.TransportService; |
||||
|
import org.thingsboard.server.common.transport.limits.TransportRateLimitService; |
||||
|
import org.thingsboard.server.common.util.ProtoUtils; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileRequestMsg; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileResponseMsg; |
||||
|
|
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.CompletableFuture; |
||||
|
import java.util.concurrent.CountDownLatch; |
||||
|
import java.util.concurrent.ExecutorService; |
||||
|
import java.util.concurrent.Executors; |
||||
|
import java.util.concurrent.Future; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.concurrent.locks.Lock; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.ArgumentMatchers.anyBoolean; |
||||
|
import static org.mockito.Mockito.doNothing; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
import static org.mockito.Mockito.times; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
|
||||
|
class DefaultTransportTenantProfileCacheTest { |
||||
|
|
||||
|
private DefaultTransportTenantProfileCache cache; |
||||
|
private TransportService transportService; |
||||
|
private TransportRateLimitService rateLimitService; |
||||
|
private ExecutorService executor; |
||||
|
|
||||
|
// Must match DefaultTransportTenantProfileCache.TENANT_PROFILE_FETCH_LOCK_STRIPES.
|
||||
|
private static final int STRIPE_COUNT = 1024; |
||||
|
|
||||
|
private final TenantId tenantA = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
// Deterministically pick a tenant that maps to a DIFFERENT stripe than tenantA, so the cross-tenant
|
||||
|
// test below cannot flake on the ~1/1024 chance two random UUIDs hash to the same stripe.
|
||||
|
private final TenantId tenantB = differentStripeFrom(tenantA); |
||||
|
|
||||
|
private static TenantId differentStripeFrom(TenantId other) { |
||||
|
Striped<Lock> probe = Striped.lock(STRIPE_COUNT); |
||||
|
TenantId candidate = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
while (probe.get(candidate) == probe.get(other)) { |
||||
|
candidate = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
} |
||||
|
return candidate; |
||||
|
} |
||||
|
|
||||
|
@BeforeEach |
||||
|
void setUp() { |
||||
|
cache = new DefaultTransportTenantProfileCache(); |
||||
|
transportService = mock(TransportService.class); |
||||
|
rateLimitService = mock(TransportRateLimitService.class); |
||||
|
doNothing().when(rateLimitService).update(any(TenantId.class), anyBoolean()); |
||||
|
cache.setTransportService(transportService); |
||||
|
cache.setRateLimitService(rateLimitService); |
||||
|
executor = Executors.newCachedThreadPool(); |
||||
|
} |
||||
|
|
||||
|
@AfterEach |
||||
|
void tearDown() { |
||||
|
executor.shutdownNow(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void fetchForOneTenantDoesNotBlockResolutionOfAnotherTenant() throws Exception { |
||||
|
CountDownLatch tenantAFetchStarted = new CountDownLatch(1); |
||||
|
CountDownLatch releaseTenantA = new CountDownLatch(1); |
||||
|
|
||||
|
GetEntityProfileResponseMsg responseA = responseFor(tenantA); |
||||
|
GetEntityProfileResponseMsg responseB = responseFor(tenantB); |
||||
|
|
||||
|
when(transportService.getEntityProfile(any())).thenAnswer(invocation -> { |
||||
|
GetEntityProfileRequestMsg msg = invocation.getArgument(0); |
||||
|
TenantId requested = TenantId.fromUUID(new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB())); |
||||
|
if (requested.equals(tenantA)) { |
||||
|
tenantAFetchStarted.countDown(); |
||||
|
releaseTenantA.await(5, TimeUnit.SECONDS); |
||||
|
return responseA; |
||||
|
} |
||||
|
return responseB; |
||||
|
}); |
||||
|
|
||||
|
// T1 starts fetching tenantA's profile and blocks inside the cross-service round-trip.
|
||||
|
Future<TenantProfile> tenantAResult = executor.submit(() -> cache.get(tenantA)); |
||||
|
assertThat(tenantAFetchStarted.await(5, TimeUnit.SECONDS)) |
||||
|
.as("tenantA fetch should have started").isTrue(); |
||||
|
|
||||
|
// T2 resolves a different tenant - it must NOT wait for tenantA's in-flight fetch.
|
||||
|
// Fails today (single global lock); passes once locking is per-tenant.
|
||||
|
TenantProfile tenantBProfile = CompletableFuture |
||||
|
.supplyAsync(() -> cache.get(tenantB), executor) |
||||
|
.get(2, TimeUnit.SECONDS); |
||||
|
assertThat(tenantBProfile).isNotNull(); |
||||
|
|
||||
|
releaseTenantA.countDown(); |
||||
|
assertThat(tenantAResult.get(5, TimeUnit.SECONDS)).isNotNull(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void concurrentMissesForSameTenantDedupeToSingleFetch() throws Exception { |
||||
|
// The per-tenant lock exists precisely so that concurrent cold misses for the SAME tenant collapse
|
||||
|
// into a single cross-service fetch (the rest are served from cache). Assert that contract directly.
|
||||
|
int callers = 8; |
||||
|
CountDownLatch fetchStarted = new CountDownLatch(1); |
||||
|
CountDownLatch releaseFetch = new CountDownLatch(1); |
||||
|
|
||||
|
when(transportService.getEntityProfile(any())).thenAnswer(invocation -> { |
||||
|
fetchStarted.countDown(); |
||||
|
// Hold the (single) in-flight fetch open while the other callers pile up on the per-tenant lock.
|
||||
|
releaseFetch.await(5, TimeUnit.SECONDS); |
||||
|
return responseFor(tenantA); |
||||
|
}); |
||||
|
|
||||
|
CountDownLatch allSubmitted = new CountDownLatch(callers); |
||||
|
List<Future<TenantProfile>> results = new ArrayList<>(); |
||||
|
for (int i = 0; i < callers; i++) { |
||||
|
results.add(executor.submit(() -> { |
||||
|
allSubmitted.countDown(); |
||||
|
return cache.get(tenantA); |
||||
|
})); |
||||
|
} |
||||
|
|
||||
|
assertThat(allSubmitted.await(5, TimeUnit.SECONDS)).as("all callers should start").isTrue(); |
||||
|
assertThat(fetchStarted.await(5, TimeUnit.SECONDS)).as("the first fetch should start").isTrue(); |
||||
|
releaseFetch.countDown(); |
||||
|
|
||||
|
for (Future<TenantProfile> result : results) { |
||||
|
assertThat(result.get(5, TimeUnit.SECONDS)).isNotNull(); |
||||
|
} |
||||
|
// All 8 callers resolved the same tenant, but only one of them hit the backend.
|
||||
|
verify(transportService, times(1)).getEntityProfile(any()); |
||||
|
} |
||||
|
|
||||
|
private GetEntityProfileResponseMsg responseFor(TenantId tenantId) { |
||||
|
TenantProfile profile = new TenantProfile(new TenantProfileId(UUID.randomUUID())); |
||||
|
profile.setName("profile-" + tenantId.getId()); |
||||
|
return GetEntityProfileResponseMsg.newBuilder() |
||||
|
.setEntityType(EntityType.TENANT.name()) |
||||
|
.setTenantProfile(ProtoUtils.toProto(profile)) |
||||
|
.setApiState(ProtoUtils.toProto(enabledApiUsageState(tenantId))) |
||||
|
.build(); |
||||
|
} |
||||
|
|
||||
|
private ApiUsageState enabledApiUsageState(TenantId tenantId) { |
||||
|
ApiUsageState state = new ApiUsageState(new ApiUsageStateId(UUID.randomUUID())); |
||||
|
state.setTenantId(tenantId); |
||||
|
state.setEntityId(tenantId); |
||||
|
state.setTransportState(ApiUsageStateValue.ENABLED); |
||||
|
state.setDbStorageState(ApiUsageStateValue.ENABLED); |
||||
|
state.setReExecState(ApiUsageStateValue.ENABLED); |
||||
|
state.setJsExecState(ApiUsageStateValue.ENABLED); |
||||
|
state.setTbelExecState(ApiUsageStateValue.ENABLED); |
||||
|
state.setEmailExecState(ApiUsageStateValue.ENABLED); |
||||
|
state.setSmsExecState(ApiUsageStateValue.ENABLED); |
||||
|
state.setAlarmExecState(ApiUsageStateValue.ENABLED); |
||||
|
state.setVersion(1L); |
||||
|
return state; |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue