Browse Source
The defective code lives in common/transport/transport-api and is shared by all transports (MQTT, HTTP, CoAP, LwM2M, SNMP); the production incident happened to surface on MQTT. On a cold tenant-profile cache (e.g. after a cache clear + restart), a device reconnect storm could serialize the whole transport instance behind tenant-profile resolution, saturating the callback pool and stalling the node for ~15 minutes. Two compounding causes are addressed: - DefaultTransportTenantProfileCache held a single process-wide ReentrantLock across the synchronous cross-service getEntityProfile round-trip, so every tenant-profile cache miss in the whole process was serialized one-at-a-time. Replace it with a bounded set of per-tenant locks (Guava Striped) so different tenants resolve concurrently while concurrent misses for the same tenant are still de-duplicated. - DefaultTransportRateLimitService performed that blocking fetch inside ConcurrentHashMap.computeIfAbsent's mapping function, holding a CHM bin lock across the remote round-trip. Pre-fetch the tenant profile before computeIfAbsent so no bin lock is held across I/O. Also de-duplicate the four near-identical getXRateLimits methods into one generic helper, move the per-type rate-limit getters onto the TransportLimitsType enum, and avoid fetching the tenant profile four times in update(TenantId).pull/15744/head
Viacheslav Klimov
4 months ago
5 changed files with 348 additions and 91 deletions
@ -0,0 +1,94 @@ |
|||
/** |
|||
* 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.thingsboard.server.common.data.TenantProfile; |
|||
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 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(); |
|||
} |
|||
|
|||
private TenantProfile tenantProfile() { |
|||
TenantProfile profile = new TenantProfile(new TenantProfileId(UUID.randomUUID())); |
|||
profile.setName("test-profile"); |
|||
TenantProfileData profileData = new TenantProfileData(); |
|||
profileData.setConfiguration(new DefaultTenantProfileConfiguration()); |
|||
profile.setProfileData(profileData); |
|||
return profile; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,136 @@ |
|||
/** |
|||
* 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 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.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 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.when; |
|||
|
|||
class DefaultTransportTenantProfileCacheTest { |
|||
|
|||
private DefaultTransportTenantProfileCache cache; |
|||
private TransportService transportService; |
|||
private TransportRateLimitService rateLimitService; |
|||
private ExecutorService executor; |
|||
|
|||
private final TenantId tenantA = TenantId.fromUUID(UUID.randomUUID()); |
|||
private final TenantId tenantB = TenantId.fromUUID(UUID.randomUUID()); |
|||
|
|||
@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(); |
|||
} |
|||
|
|||
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