diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/service/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/service/EdgeGrpcService.java index b50d643634..55278d7f8b 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/service/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/service/EdgeGrpcService.java @@ -64,12 +64,16 @@ import org.thingsboard.server.service.edge.rpc.session.EdgeSessionsHolder; import org.thingsboard.server.service.edge.rpc.session.manager.EdgeGrpcSessionManager; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; +import java.util.ArrayList; +import java.util.List; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE; @@ -89,6 +93,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @Value("${edges.state.persistToTelemetry:false}") private boolean persistToTelemetry; + @Value("${edges.connectivity.disconnect_notification_delay_ms:60000}") + private long disconnectNotificationDelayMs; + @Autowired @Lazy private EdgeContextComponent ctx; @@ -100,18 +107,24 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private final TelemetrySubscriptionService tsSubService; private final TbTransactionalCache edgeIdServiceIdCache; + private final ConcurrentMap pendingDisconnectNotifications = new ConcurrentHashMap<>(); private final ConcurrentMap> localSyncEdgeRequests = new ConcurrentHashMap<>(); private ScheduledExecutorService executorService; private ScheduledExecutorService sendDownlinkExecutorService; @PostConstruct public void onStartUp() { - this.executorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("edge-service"); + // removeOnCancelPolicy: cancelled delayed disconnect notifications (common with flapping edges) are dropped + // from the queue right away on reconnect, instead of piling up until their original fire time. + this.executorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("edge-service", true); this.sendDownlinkExecutorService = ThingsBoardExecutors.newScheduledThreadPool(sendSchedulerPoolSize, "edge-send-scheduler"); } @PreDestroy private void preDestroy() { + List pendingToFlush = new ArrayList<>(pendingDisconnectNotifications.values()); + pendingDisconnectNotifications.clear(); + pendingToFlush.forEach(this::fireDelayedDisconnectNotification); shutdownExecutorSafely(executorService); shutdownExecutorSafely(sendDownlinkExecutorService); sessions.forEach(EdgeGrpcSessionManager::onEdgeDisconnect); @@ -210,7 +223,12 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i long lastConnectTs = System.currentTimeMillis(); save(tenantId, edgeId, LAST_CONNECT_TIME, lastConnectTs); edgeIdServiceIdCache.put(edgeId, serviceInfoProvider.getServiceId()); + // 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); pushRuleEngineMessage(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); + // Connect notifies immediately; only the disconnect notification is debounced (see scheduleDisconnectNotification). + notifyEdgeConnectivity(edge, true); edgeSession.onEdgeConnect(); } @@ -230,6 +248,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i long lastDisconnectTs = System.currentTimeMillis(); save(tenantId, edgeId, LAST_DISCONNECT_TIME, lastDisconnectTs); pushRuleEngineMessage(tenantId, edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); + scheduleDisconnectNotification(edge); } else { log.info("[{}] edge session [{}] is not current anymore. Attempting to destroy it by sessionId.", edgeId, sessionId); EdgeGrpcSessionManager stale = sessions.removeBySessionId(sessionId); @@ -318,12 +337,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) { @@ -344,6 +357,79 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } + private void notifyEdgeConnectivity(Edge edge, boolean connected) { + try { + ctx.getRuleProcessor().process(EdgeConnectionTrigger.builder() + .tenantId(edge.getTenantId()) + .customerId(edge.getCustomerId()) + .edgeId(edge.getId()) + .edgeName(edge.getName()) + .connected(connected).build()); + } catch (Exception e) { + log.warn("[{}][{}] Failed to process edge connectivity notification (connected={})", edge.getTenantId(), edge.getId(), connected, e); + } + } + + private void scheduleDisconnectNotification(Edge edge) { + // Zero delay means "no debounce": notify immediately and skip the cluster-wide re-verify guard. + if (disconnectNotificationDelayMs <= 0) { + notifyEdgeConnectivity(edge, false); + return; + } + EdgeId edgeId = edge.getId(); + pendingDisconnectNotifications.compute(edgeId, (id, existing) -> { + if (existing != null) { + cancelIfPending(existing.getFuture()); + } + // Create the pending entry first so the scheduled task can reference it (for the identity-keyed + // remove in fireDelayedDisconnectNotification), then back-fill its future. Doing this inside compute() + // keeps it under the map bin lock, so the future is set before the entry becomes visible to other threads. + PendingDisconnect pending = new PendingDisconnect(edge); + ScheduledFuture future = executorService.schedule( + () -> fireDelayedDisconnectNotification(pending), + disconnectNotificationDelayMs, TimeUnit.MILLISECONDS); + pending.setFuture(future); + return pending; + }); + } + + private void fireDelayedDisconnectNotification(PendingDisconnect pending) { + // Claim-once guard: the @PreDestroy flush and a concurrently-firing scheduled task can both call this for + // the same PendingDisconnect. tryClaim() ensures notifyEdgeConnectivity fires at most once without relying + // on downstream notification dedup. + if (!pending.tryClaim()) { + return; + } + Edge edge = pending.getEdge(); + EdgeId edgeId = edge.getId(); + // Identity-keyed remove: don't clobber a newer entry if a second disconnect races with this task firing. + pendingDisconnectNotifications.remove(edgeId, pending); + // 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.hasByEdgeId(edgeId) || isConnectedClusterWide(edgeId)) { + log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", edge.getTenantId(), edgeId); + return; + } + notifyEdgeConnectivity(edge, false); + } + + private void cancelPendingDisconnectNotification(EdgeId edgeId) { + PendingDisconnect pending = pendingDisconnectNotifications.remove(edgeId); + if (pending != null) { + cancelIfPending(pending.getFuture()); + } + } + + private boolean isConnectedClusterWide(EdgeId edgeId) { + return edgeIdServiceIdCache.get(edgeId) != null; + } + + private static void cancelIfPending(ScheduledFuture future) { + if (future != null && !future.isDone()) { + future.cancel(false); + } + } + private void save(TenantId tenantId, EdgeId edgeId, String key, long value) { log.debug("[{}][{}] Updating long edge telemetry [{}] [{}]", tenantId, edgeId, key, value); if (persistToTelemetry) { @@ -389,4 +475,33 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i e.shutdown(); } } + + static final class PendingDisconnect { + + private final Edge edge; + private final AtomicBoolean notified = new AtomicBoolean(false); + private ScheduledFuture future; + + PendingDisconnect(Edge edge) { + this.edge = edge; + } + + Edge getEdge() { + return edge; + } + + ScheduledFuture getFuture() { + return future; + } + + void setFuture(ScheduledFuture future) { + this.future = future; + } + + boolean tryClaim() { + return notified.compareAndSet(false, 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 index 4482177750..b410b5ca43 100644 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -28,7 +28,7 @@ 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 org.thingsboard.server.service.edge.rpc.service.EdgeGrpcService; import java.util.Map; import java.util.concurrent.TimeUnit;