Viacheslav Klimov
2 days ago
committed by
GitHub
GitHub
9 changed files with 418 additions and 5 deletions
@ -0,0 +1,133 @@ |
|||
/** |
|||
* 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.edge; |
|||
|
|||
import org.junit.After; |
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.mockito.ArgumentMatcher; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTrigger; |
|||
import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; |
|||
import org.thingsboard.server.controller.AbstractWebTest; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.edge.imitator.EdgeImitator; |
|||
import org.thingsboard.server.service.edge.rpc.EdgeGrpcService; |
|||
|
|||
import java.util.Map; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.awaitility.Awaitility.await; |
|||
import static org.mockito.ArgumentMatchers.argThat; |
|||
import static org.mockito.Mockito.atLeastOnce; |
|||
import static org.mockito.Mockito.clearInvocations; |
|||
import static org.mockito.Mockito.never; |
|||
import static org.mockito.Mockito.times; |
|||
import static org.mockito.Mockito.verify; |
|||
|
|||
@DaoSqlTest |
|||
public class EdgeConnectionNotificationTest extends AbstractEdgeTest { |
|||
|
|||
private static final long DELAY_MS = 1500L; |
|||
|
|||
@MockitoSpyBean |
|||
private NotificationRuleProcessor notificationRuleProcessor; |
|||
|
|||
@Autowired |
|||
private EdgeGrpcService edgeGrpcService; |
|||
|
|||
private long originalDisconnectNotificationDelayMs; |
|||
|
|||
// Capture the bean's configured delay before each test and restore it after, so a test method that forgets
|
|||
// to call setDisconnectNotificationDelayMs can't silently inherit the previous method's mutated value
|
|||
// (the shared context means the mutation would otherwise persist across methods).
|
|||
@Before |
|||
public void captureDisconnectNotificationDelay() { |
|||
originalDisconnectNotificationDelayMs = (long) ReflectionTestUtils.getField(edgeGrpcService, "disconnectNotificationDelayMs"); |
|||
} |
|||
|
|||
@After |
|||
public void restoreDisconnectNotificationDelay() { |
|||
setDisconnectNotificationDelayMs(originalDisconnectNotificationDelayMs); |
|||
} |
|||
|
|||
// The delay is overridden per test (rather than via a per-class @TestPropertySource) so all cases share a
|
|||
// single Spring application context instead of booting a separate heavy context per delay value.
|
|||
private void setDisconnectNotificationDelayMs(long delayMs) { |
|||
ReflectionTestUtils.setField(edgeGrpcService, "disconnectNotificationDelayMs", delayMs); |
|||
} |
|||
|
|||
@Test |
|||
public void givenEdgeStaysDisconnected_whenDelayElapses_thenDisconnectNotificationSent() throws Exception { |
|||
// After the configured delay, the "disconnected" notification is sent exactly once.
|
|||
assertDisconnectNotificationSentOnce(DELAY_MS); |
|||
} |
|||
|
|||
@Test |
|||
public void givenZeroDelay_whenEdgeDisconnects_thenDisconnectNotificationSentImmediately() throws Exception { |
|||
// With a zero delay there is no debounce window - the "disconnected" notification fires right away.
|
|||
assertDisconnectNotificationSentOnce(0); |
|||
} |
|||
|
|||
private void assertDisconnectNotificationSentOnce(long delayMs) throws Exception { |
|||
setDisconnectNotificationDelayMs(delayMs); |
|||
clearInvocations(notificationRuleProcessor); |
|||
|
|||
edgeImitator.disconnect(); |
|||
|
|||
await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> |
|||
verify(notificationRuleProcessor, times(1)).process(argThat(edgeConnectionTrigger(false)))); |
|||
} |
|||
|
|||
@Test |
|||
public void givenEdgeReconnectsWithinDelay_whenEdgeFlaps_thenDisconnectNotificationSuppressed() throws Exception { |
|||
setDisconnectNotificationDelayMs(DELAY_MS); |
|||
clearInvocations(notificationRuleProcessor); |
|||
|
|||
// Edge drops...
|
|||
edgeImitator.disconnect(); |
|||
// Wait until the server has processed the disconnect and scheduled the pending notification
|
|||
// (the edge id appears in the pendingDisconnectNotifications map) before reconnecting.
|
|||
await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).until(() -> { |
|||
Map<?, ?> pending = (Map<?, ?>) ReflectionTestUtils.getField(edgeGrpcService, "pendingDisconnectNotifications"); |
|||
return pending != null && pending.containsKey(edge.getId()); |
|||
}); |
|||
|
|||
// ...and reconnects within the delay window, which must cancel the pending "disconnected" notification.
|
|||
EdgeImitator reconnected = createEdgeImitator(); |
|||
reconnected.connect(); |
|||
edgeImitator = reconnected; // let teardown clean up the live session
|
|||
|
|||
// The "connected" notification still fires immediately on reconnect (we suppress the disconnect only).
|
|||
await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> |
|||
verify(notificationRuleProcessor, atLeastOnce()).process(argThat(edgeConnectionTrigger(true)))); |
|||
|
|||
// The "disconnected" notification must never be sent throughout the full delay window.
|
|||
await().during(DELAY_MS + 500, TimeUnit.MILLISECONDS) |
|||
.atMost(DELAY_MS + 2000, TimeUnit.MILLISECONDS) |
|||
.untilAsserted(() -> verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false)))); |
|||
} |
|||
|
|||
private ArgumentMatcher<NotificationRuleTrigger> edgeConnectionTrigger(boolean connected) { |
|||
return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger |
|||
&& edge.getId().equals(edgeTrigger.getEdgeId()) |
|||
&& edgeTrigger.isConnected() == connected; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,168 @@ |
|||
/** |
|||
* 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.edge.rpc; |
|||
|
|||
import io.grpc.Server; |
|||
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; |
|||
import io.grpc.netty.shaded.io.netty.buffer.PooledByteBufAllocator; |
|||
import io.grpc.stub.StreamObserver; |
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.gen.edge.v1.ConnectResponseCode; |
|||
import org.thingsboard.server.gen.edge.v1.ConnectResponseMsg; |
|||
import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc; |
|||
import org.thingsboard.server.gen.edge.v1.RequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RequestMsgType; |
|||
import org.thingsboard.server.gen.edge.v1.ResponseMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UplinkMsg; |
|||
|
|||
import java.lang.reflect.Method; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.BooleanSupplier; |
|||
|
|||
import static org.junit.jupiter.api.Assertions.fail; |
|||
|
|||
class EdgeGrpcClientLeakTest { |
|||
|
|||
// The window in which the default shared event loop group would be destroyed after the channel
|
|||
// terminates (SharedResourceHolder delays destruction by 1 second). If EdgeGrpcClient ever goes
|
|||
// back to the shared group, writes after this window hit a terminated executor and every buffer
|
|||
// committed to grpc-netty's WriteQueue is pinned forever (4112 bytes per message, silent after
|
|||
// the first RejectedExecutionException).
|
|||
private static final long SHARED_GROUP_DEATH_WINDOW_MS = 3000; |
|||
private static final int MSG_COUNT = 50; |
|||
private static final long AWAIT_TIMEOUT_MS = 15_000; |
|||
|
|||
private Server server; |
|||
private EdgeGrpcClient client; |
|||
|
|||
@Test |
|||
void uplinksSentAfterTransportDeathDoNotPinPooledBuffers() throws Exception { |
|||
server = NettyServerBuilder.forPort(0) |
|||
.addService(new EdgeRpcServiceGrpc.EdgeRpcServiceImplBase() { |
|||
@Override |
|||
public StreamObserver<RequestMsg> handleMsgs(StreamObserver<ResponseMsg> outputStream) { |
|||
return new StreamObserver<>() { |
|||
@Override |
|||
public void onNext(RequestMsg requestMsg) { |
|||
if (requestMsg.hasConnectRequestMsg()) { |
|||
outputStream.onNext(ResponseMsg.newBuilder() |
|||
.setConnectResponseMsg(ConnectResponseMsg.newBuilder() |
|||
.setResponseCode(ConnectResponseCode.ACCEPTED) |
|||
.build()) |
|||
.build()); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onError(Throwable t) { |
|||
} |
|||
|
|||
@Override |
|||
public void onCompleted() { |
|||
} |
|||
}; |
|||
} |
|||
}) |
|||
.build() |
|||
.start(); |
|||
|
|||
client = new EdgeGrpcClient(); |
|||
ReflectionTestUtils.setField(client, "rpcHost", "localhost"); |
|||
ReflectionTestUtils.setField(client, "rpcPort", server.getPort()); |
|||
ReflectionTestUtils.setField(client, "timeoutSecs", 1); |
|||
ReflectionTestUtils.setField(client, "keepAliveTimeSec", 10); |
|||
ReflectionTestUtils.setField(client, "keepAliveTimeoutSec", 5); |
|||
ReflectionTestUtils.setField(client, "maxInboundMessageSize", 4194304); |
|||
|
|||
client.connect("leakTest", "leakTest", msg -> {}, cfg -> {}, msg -> {}, e -> {}); |
|||
await("client to connect", () -> client.isConnected()); |
|||
|
|||
server.shutdownNow(); |
|||
server.awaitTermination(10, TimeUnit.SECONDS); |
|||
await("client to observe the transport death", () -> !client.isConnected()); |
|||
Thread.sleep(SHARED_GROUP_DEATH_WINDOW_MS); |
|||
|
|||
long baseline = pinnedBytes(); |
|||
@SuppressWarnings("unchecked") |
|||
StreamObserver<RequestMsg> inputStream = (StreamObserver<RequestMsg>) ReflectionTestUtils.getField(client, "inputStream"); |
|||
RequestMsg uplink = RequestMsg.newBuilder() |
|||
.setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) |
|||
.setUplinkMsg(UplinkMsg.newBuilder().setUplinkMsgId(1).build()) |
|||
.build(); |
|||
// Bypasses the connected gate on purpose: this models the check-then-act straggler (and the
|
|||
// pre-gate retry loop) writing to a stream whose transport is already gone. Exceptions are
|
|||
// swallowed the same way the production retry loop survives them.
|
|||
for (int i = 0; i < MSG_COUNT; i++) { |
|||
try { |
|||
inputStream.onNext(uplink); |
|||
} catch (RuntimeException ignored) { |
|||
} |
|||
} |
|||
|
|||
long deadline = System.currentTimeMillis() + AWAIT_TIMEOUT_MS; |
|||
while (pinnedBytes() > baseline) { |
|||
if (System.currentTimeMillis() > deadline) { |
|||
fail("Pinned pooled memory did not return to baseline: " + (pinnedBytes() - baseline) |
|||
+ " bytes retained after " + MSG_COUNT + " uplinks to a dead stream"); |
|||
} |
|||
Thread.sleep(50); |
|||
} |
|||
} |
|||
|
|||
@AfterEach |
|||
void tearDown() throws Exception { |
|||
if (client != null) { |
|||
client.disconnect(true); |
|||
client.destroy(); |
|||
} |
|||
if (server != null) { |
|||
server.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
private void await(String what, BooleanSupplier condition) throws InterruptedException { |
|||
long deadline = System.currentTimeMillis() + AWAIT_TIMEOUT_MS; |
|||
while (!condition.getAsBoolean()) { |
|||
if (System.currentTimeMillis() > deadline) { |
|||
fail("Timed out waiting for " + what); |
|||
} |
|||
Thread.sleep(50); |
|||
} |
|||
} |
|||
|
|||
// grpc-netty builds its own PooledByteBufAllocator instances instead of using
|
|||
// PooledByteBufAllocator.DEFAULT, and the factory that owns them is package private - so they
|
|||
// have to be pulled out reflectively. Both variants are checked so the assertion holds no matter
|
|||
// which one the transport picks on this platform/version.
|
|||
private static long pinnedBytes() { |
|||
try { |
|||
Class<?> utils = Class.forName("io.grpc.netty.shaded.io.grpc.netty.Utils"); |
|||
Method getByteBufAllocator = utils.getDeclaredMethod("getByteBufAllocator", boolean.class); |
|||
getByteBufAllocator.setAccessible(true); |
|||
long total = 0; |
|||
for (boolean forceHeapBuffer : new boolean[]{false, true}) { |
|||
PooledByteBufAllocator pooled = (PooledByteBufAllocator) getByteBufAllocator.invoke(null, forceHeapBuffer); |
|||
total += pooled.pinnedDirectMemory() + pooled.pinnedHeapMemory(); |
|||
} |
|||
return total; |
|||
} catch (Exception e) { |
|||
throw new IllegalStateException("Failed to read the gRPC allocator metrics", e); |
|||
} |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue