From 361e267db4aeddc9b791b69e48fe2798bed4f670 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Fri, 5 Jun 2026 15:06:55 +0300 Subject: [PATCH] Edge: prevent disconnect notification spam from flapping edges --- .../service/edge/rpc/EdgeGrpcService.java | 104 ++++++++++++++++-- .../src/main/resources/thingsboard.yml | 5 + .../edge/EdgeConnectionNotificationTest.java | 96 ++++++++++++++++ 3 files changed, 194 insertions(+), 11 deletions(-) create mode 100644 application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index 072a4878d5..7f57101205 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -41,7 +41,6 @@ import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.StringUtils; -import org.thingsboard.server.common.transport.config.ssl.PemSslCredentials; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.id.EdgeId; @@ -58,6 +57,7 @@ import org.thingsboard.server.common.msg.edge.EdgeHighPriorityMsg; import org.thingsboard.server.common.msg.edge.EdgeSessionMsg; import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse; import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest; +import org.thingsboard.server.common.transport.config.ssl.PemSslCredentials; import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc; import org.thingsboard.server.gen.edge.v1.RequestMsg; import org.thingsboard.server.gen.edge.v1.ResponseMsg; @@ -68,7 +68,6 @@ import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.edge.EdgeContextComponent; -import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import java.io.IOException; import java.util.ArrayList; @@ -103,6 +102,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private final ConcurrentMap sessionNewEventsLocks = new ConcurrentHashMap<>(); private final Map sessionNewEvents = new HashMap<>(); private final ConcurrentMap> sessionEdgeEventChecks = new ConcurrentHashMap<>(); + private final ConcurrentMap> pendingDisconnectNotifications = new ConcurrentHashMap<>(); private final ConcurrentMap> localSyncEdgeRequests = new ConcurrentHashMap<>(); private final ConcurrentMap edgeEventsMigrationProcessed = new ConcurrentHashMap<>(); private final List zombieSessions = new ArrayList<>(); @@ -136,6 +136,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @Value("${edges.max_high_priority_queue_size_per_session:10000}") private int maxHighPriorityQueueSizePerSession; + @Value("${edges.connectivity.disconnect_notification_delay_ms:60000}") + private long disconnectNotificationDelayMs; + @Autowired @Lazy private EdgeContextComponent ctx; @@ -239,6 +242,12 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i sessionEdgeEventChecks.remove(edgeId); } } + pendingDisconnectNotifications.values().forEach(task -> { + if (task != null && !task.isDone()) { + task.cancel(false); + } + }); + pendingDisconnectNotifications.clear(); if (edgeEventProcessingExecutorService != null) { edgeEventProcessingExecutorService.shutdownNow(); } @@ -321,6 +330,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } finally { newEventLock.unlock(); } + cancelPendingDisconnectNotification(edgeId); cancelScheduleEdgeEventsCheck(edgeId); } } @@ -383,7 +393,11 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i long lastConnectTs = System.currentTimeMillis(); save(tenantId, edgeId, LAST_CONNECT_TIME, lastConnectTs); edgeIdServiceIdCache.put(edgeId, serviceInfoProvider.getServiceId()); - pushRuleEngineMessage(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); + // If the edge reconnected within the disconnect-notification delay window, suppress the pending + // "disconnected" notification - the drop was transient (debounce for flapping edges). + cancelPendingDisconnectNotification(edgeId); + pushStateEventToRuleEngine(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); + notifyEdgeConnectivity(tenantId, edge, true); cancelScheduleEdgeEventsCheck(edgeId); edgeEventsMigrationProcessed.putIfAbsent(edgeId, Boolean.FALSE); scheduleEdgeEventsCheck(edgeGrpcSession); @@ -545,7 +559,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i save(tenantId, edgeId, ACTIVITY_STATE, false); long lastDisconnectTs = System.currentTimeMillis(); save(tenantId, edgeId, LAST_DISCONNECT_TIME, lastDisconnectTs); - pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); + pushStateEventToRuleEngine(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); + scheduleDisconnectNotification(tenantId, edge); cancelScheduleEdgeEventsCheck(edgeId); } else { log.info("[{}] edge session [{}] is not current anymore. Attempting to destroy it by sessionId.", edgeId, sessionId); @@ -617,7 +632,33 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } - private void pushRuleEngineMessage(TenantId tenantId, Edge edge, long ts, TbMsgType msgType) { + private static class AttributeSaveCallback implements FutureCallback { + + private final TenantId tenantId; + private final EdgeId edgeId; + private final String key; + private final Object value; + + AttributeSaveCallback(TenantId tenantId, EdgeId edgeId, String key, Object value) { + this.tenantId = tenantId; + this.edgeId = edgeId; + this.key = key; + this.value = value; + } + + @Override + public void onSuccess(@Nullable Void result) { + log.trace("[{}][{}] Successfully updated attribute [{}] with value [{}]", tenantId, edgeId, key, value); + } + + @Override + public void onFailure(Throwable t) { + log.warn("[{}][{}] Failed to update attribute [{}] with value [{}]", tenantId, edgeId, key, value, t); + } + + } + + private void pushStateEventToRuleEngine(TenantId tenantId, Edge edge, long ts, TbMsgType msgType) { try { EdgeId edgeId = edge.getId(); ObjectNode edgeState = JacksonUtil.newObjectNode(); @@ -629,12 +670,6 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i edgeState.put(ACTIVITY_STATE, false); edgeState.put(LAST_DISCONNECT_TIME, ts); } - ctx.getRuleProcessor().process(EdgeConnectionTrigger.builder() - .tenantId(tenantId) - .customerId(edge.getCustomerId()) - .edgeId(edgeId) - .edgeName(edge.getName()) - .connected(isConnected).build()); String data = JacksonUtil.toString(edgeState); TbMsgMetaData md = new TbMsgMetaData(); if (!persistToTelemetry) { @@ -655,6 +690,53 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } + private void notifyEdgeConnectivity(TenantId tenantId, Edge edge, boolean connected) { + try { + ctx.getRuleProcessor().process(EdgeConnectionTrigger.builder() + .tenantId(tenantId) + .customerId(edge.getCustomerId()) + .edgeId(edge.getId()) + .edgeName(edge.getName()) + .connected(connected).build()); + } catch (Exception e) { + log.warn("[{}][{}] Failed to process edge connectivity notification (connected={})", tenantId, edge.getId(), connected, e); + } + } + + private void scheduleDisconnectNotification(TenantId tenantId, Edge edge) { + if (disconnectNotificationDelayMs <= 0) { + notifyEdgeConnectivity(tenantId, edge, false); + return; + } + EdgeId edgeId = edge.getId(); + pendingDisconnectNotifications.compute(edgeId, (id, existing) -> { + if (existing != null && !existing.isDone()) { + existing.cancel(false); + } + return executorService.schedule(() -> fireDelayedDisconnectNotification(tenantId, edge), + disconnectNotificationDelayMs, TimeUnit.MILLISECONDS); + }); + } + + private void fireDelayedDisconnectNotification(TenantId tenantId, Edge edge) { + EdgeId edgeId = edge.getId(); + pendingDisconnectNotifications.remove(edgeId); + // Re-verify the edge is still disconnected. The cache is cluster-wide, so this also covers the case + // where the edge dropped on this node and reconnected to a different TB-Core node within the window. + if (sessions.containsKey(edgeId) || edgeIdServiceIdCache.get(edgeId) != null) { + log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", tenantId, edgeId); + return; + } + notifyEdgeConnectivity(tenantId, edge, false); + } + + private void cancelPendingDisconnectNotification(EdgeId edgeId) { + ScheduledFuture pending = pendingDisconnectNotifications.remove(edgeId); + if (pending != null && !pending.isDone()) { + pending.cancel(false); + } + } + private void cleanupZombieSessions() { try { tryToDestroyZombieSessions(getZombieSessions(sessions.values()), s -> sessions.remove(s.getEdge().getId())); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 454c59d7d7..0cb3e57ae8 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1642,6 +1642,11 @@ edges: state: # Persist state of edge (active, last connect, last disconnect) into timeseries or attributes tables. 'false' means to store edge state into attributes table persistToTelemetry: "${EDGES_PERSIST_STATE_TO_TELEMETRY:false}" + connectivity: + # Delay (ms) before sending an edge "disconnected" notification. Suppressed if the edge reconnects within + # this window - debounces flapping edges from spamming the notification center. Only the notification is + # delayed; rule-engine events and state attributes update immediately. Set to 0 to notify immediately. + disconnect_notification_delay_ms: "${TB_EDGE_DISCONNECT_NOTIFICATION_DELAY_MS:60000}" stats: # Enable or disable reporting of edge communication stats (true or false) enabled: "${EDGES_STATS_ENABLED:true}" diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java new file mode 100644 index 0000000000..f72ad51f4a --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -0,0 +1,96 @@ +/** + * 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.Test; +import org.mockito.ArgumentMatcher; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +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.gen.edge.v1.OAuth2ClientUpdateMsg; +import org.thingsboard.server.gen.edge.v1.OAuth2DomainUpdateMsg; + +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 +@TestPropertySource(properties = { + "edges.connectivity.disconnect_notification_delay_ms=5000" +}) +public class EdgeConnectionNotificationTest extends AbstractEdgeTest { + + private static final long DELAY_MS = 5000L; + + @MockitoSpyBean + private NotificationRuleProcessor notificationRuleProcessor; + + @Test + public void givenEdgeStaysDisconnected_whenDelayElapses_thenDisconnectNotificationSent() throws Exception { + clearInvocations(notificationRuleProcessor); + + edgeImitator.disconnect(); + + // After the configured delay, the "disconnected" notification is sent exactly once. + await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> + verify(notificationRuleProcessor, times(1)).process(argThat(edgeConnectionTrigger(false)))); + } + + @Test + public void givenEdgeReconnectsWithinDelay_whenEdgeFlaps_thenDisconnectNotificationSuppressed() throws Exception { + clearInvocations(notificationRuleProcessor); + + // Edge drops... + edgeImitator.disconnect(); + // Ensure the server processed the disconnect (and scheduled the delayed notification) before reconnecting. + TimeUnit.SECONDS.sleep(1); + + // ...and reconnects within the delay window, which must cancel the pending "disconnected" notification. + EdgeImitator reconnected = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); + reconnected.ignoreType(OAuth2ClientUpdateMsg.class); + reconnected.ignoreType(OAuth2DomainUpdateMsg.class); + 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)))); + + // Wait until the original disconnect-notification window has fully elapsed... + TimeUnit.MILLISECONDS.sleep(DELAY_MS + 1000); + + // ...the "disconnected" notification must have never been sent. + verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false))); + } + + private ArgumentMatcher edgeConnectionTrigger(boolean connected) { + return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger + && edge.getId().equals(edgeTrigger.getEdgeId()) + && edgeTrigger.isConnected() == connected; + } + +}