Viacheslav Klimov
3 days ago
committed by
GitHub
GitHub
6 changed files with 538 additions and 35 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,194 @@ |
|||||
|
/** |
||||
|
* 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.service.edge.rpc; |
||||
|
|
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.junit.jupiter.api.extension.ExtendWith; |
||||
|
import org.mockito.ArgumentMatcher; |
||||
|
import org.mockito.InjectMocks; |
||||
|
import org.mockito.Mock; |
||||
|
import org.mockito.junit.jupiter.MockitoExtension; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
import org.thingsboard.server.cache.SimpleTbCacheValueWrapper; |
||||
|
import org.thingsboard.server.cache.TbTransactionalCache; |
||||
|
import org.thingsboard.server.common.data.edge.Edge; |
||||
|
import org.thingsboard.server.common.data.id.EdgeId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
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.queue.discovery.TbServiceInfoProvider; |
||||
|
import org.thingsboard.server.service.edge.EdgeContextComponent; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.ConcurrentMap; |
||||
|
|
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.ArgumentMatchers.argThat; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
import static org.mockito.Mockito.never; |
||||
|
import static org.mockito.Mockito.times; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
|
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
public class EdgeGrpcServiceTest { |
||||
|
|
||||
|
private static final String THIS_NODE = "tb-core-1"; |
||||
|
private static final String OTHER_NODE = "tb-core-2"; |
||||
|
|
||||
|
@Mock |
||||
|
private TbTransactionalCache<EdgeId, String> edgeIdServiceIdCache; |
||||
|
|
||||
|
@Mock |
||||
|
private TbServiceInfoProvider serviceInfoProvider; |
||||
|
|
||||
|
@Mock |
||||
|
private EdgeContextComponent ctx; |
||||
|
|
||||
|
@Mock |
||||
|
private NotificationRuleProcessor ruleProcessor; |
||||
|
|
||||
|
@InjectMocks |
||||
|
private EdgeGrpcService edgeGrpcService; |
||||
|
|
||||
|
private EdgeId edgeId; |
||||
|
private Edge edge; |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setUp() { |
||||
|
edgeId = new EdgeId(UUID.randomUUID()); |
||||
|
edge = new Edge(edgeId); |
||||
|
edge.setTenantId(TenantId.fromUUID(UUID.randomUUID())); |
||||
|
edge.setName("test-edge"); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenCacheOwnedByThisNode_whenEvict_thenEntryIsEvicted() { |
||||
|
when(serviceInfoProvider.getServiceId()).thenReturn(THIS_NODE); |
||||
|
when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(THIS_NODE)); |
||||
|
|
||||
|
evictServiceIdCacheIfOwnedByThisNode(); |
||||
|
|
||||
|
verify(edgeIdServiceIdCache, times(1)).evict(edgeId); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenCacheOwnedByAnotherNode_whenEvict_thenEntryIsKept() { |
||||
|
// The edge already reconnected to another node within the keep-alive window: must NOT wipe the live owner.
|
||||
|
when(serviceInfoProvider.getServiceId()).thenReturn(THIS_NODE); |
||||
|
when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); |
||||
|
|
||||
|
evictServiceIdCacheIfOwnedByThisNode(); |
||||
|
|
||||
|
verify(edgeIdServiceIdCache, never()).evict(edgeId); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenEmptyCache_whenEvict_thenNothingEvicted() { |
||||
|
when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); |
||||
|
|
||||
|
evictServiceIdCacheIfOwnedByThisNode(); |
||||
|
|
||||
|
verify(edgeIdServiceIdCache, never()).evict(edgeId); |
||||
|
} |
||||
|
|
||||
|
// --- fireDelayedDisconnectNotification re-verify guard ---
|
||||
|
|
||||
|
@Test |
||||
|
public void givenEdgeReconnectedToThisNode_whenDelayFires_thenNotificationSuppressed() { |
||||
|
// A live session exists again on this node - suppress the stale disconnect notification.
|
||||
|
sessions().put(edgeId, mock(EdgeGrpcSession.class)); |
||||
|
|
||||
|
fireDelayedDisconnectNotification(); |
||||
|
|
||||
|
verify(ruleProcessor, never()).process(any()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenEdgeReconnectedToAnotherNode_whenDelayFires_thenNotificationSuppressed() { |
||||
|
// No local session, but the cluster cache still points at some node: the edge is connected elsewhere.
|
||||
|
when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); |
||||
|
|
||||
|
fireDelayedDisconnectNotification(); |
||||
|
|
||||
|
verify(ruleProcessor, never()).process(any()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenEdgeStaysDisconnectedClusterWide_whenDelayFires_thenNotificationSent() { |
||||
|
// No local session and no cache entry on any node: the edge is genuinely down - fire the notification.
|
||||
|
when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); |
||||
|
when(ctx.getRuleProcessor()).thenReturn(ruleProcessor); |
||||
|
|
||||
|
fireDelayedDisconnectNotification(); |
||||
|
|
||||
|
verify(ruleProcessor, times(1)).process(argThat(disconnectTrigger())); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenPendingDisconnect_whenDestroy_thenNotificationFlushed() { |
||||
|
// The edge is genuinely down (no session, no cache). A graceful shutdown must flush the pending
|
||||
|
// notification rather than drop it, otherwise a restart within the delay window swallows the alert.
|
||||
|
when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); |
||||
|
when(ctx.getRuleProcessor()).thenReturn(ruleProcessor); |
||||
|
pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(edge)); |
||||
|
|
||||
|
destroy(); |
||||
|
|
||||
|
verify(ruleProcessor, times(1)).process(argThat(disconnectTrigger())); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenPendingDisconnectButReconnectedElsewhere_whenDestroy_thenNotificationSuppressed() { |
||||
|
// The flush still honors the re-verify guard: an edge that reconnected to another node must not alert.
|
||||
|
when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); |
||||
|
pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(edge)); |
||||
|
|
||||
|
destroy(); |
||||
|
|
||||
|
verify(ruleProcessor, never()).process(any()); |
||||
|
} |
||||
|
|
||||
|
private void destroy() { |
||||
|
ReflectionTestUtils.invokeMethod(edgeGrpcService, "destroy"); |
||||
|
} |
||||
|
|
||||
|
private void evictServiceIdCacheIfOwnedByThisNode() { |
||||
|
ReflectionTestUtils.invokeMethod(edgeGrpcService, "evictServiceIdCacheIfOwnedByThisNode", edgeId); |
||||
|
} |
||||
|
|
||||
|
private void fireDelayedDisconnectNotification() { |
||||
|
EdgeGrpcService.PendingDisconnect pending = new EdgeGrpcService.PendingDisconnect(edge); |
||||
|
ReflectionTestUtils.invokeMethod(edgeGrpcService, "fireDelayedDisconnectNotification", pending); |
||||
|
} |
||||
|
|
||||
|
@SuppressWarnings("unchecked") |
||||
|
private ConcurrentMap<EdgeId, EdgeGrpcSession> sessions() { |
||||
|
return (ConcurrentMap<EdgeId, EdgeGrpcSession>) ReflectionTestUtils.getField(edgeGrpcService, "sessions"); |
||||
|
} |
||||
|
|
||||
|
@SuppressWarnings("unchecked") |
||||
|
private ConcurrentMap<EdgeId, EdgeGrpcService.PendingDisconnect> pendingDisconnects() { |
||||
|
return (ConcurrentMap<EdgeId, EdgeGrpcService.PendingDisconnect>) ReflectionTestUtils.getField(edgeGrpcService, "pendingDisconnectNotifications"); |
||||
|
} |
||||
|
|
||||
|
private static ArgumentMatcher<NotificationRuleTrigger> disconnectTrigger() { |
||||
|
return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger && !edgeTrigger.isConnected(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue