From 3860d79613fb3b34f186558c9e4cbd246f22c270 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Mon, 10 Feb 2025 18:42:25 +0200 Subject: [PATCH] Improved Edge handling in case connect/disconnect. Speed up initial setup --- .../instructions/upgrade/docker/upgrade_db.md | 17 ++----------- .../service/edge/rpc/EdgeGrpcService.java | 3 +++ .../service/edge/rpc/EdgeGrpcSession.java | 8 ++++-- .../dashboard/BaseDashboardProcessor.java | 5 +++- .../edge/rpc/sync/EdgeRequestsService.java | 5 ++++ .../server/edge/DashboardEdgeTest.java | 25 ++++++++++++++++--- .../provider/KafkaMonolithQueueFactory.java | 2 +- .../provider/KafkaTbCoreQueueFactory.java | 2 +- 8 files changed, 43 insertions(+), 24 deletions(-) diff --git a/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md b/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md index a594ebe4e7..386c5f2743 100644 --- a/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md +++ b/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md @@ -43,19 +43,6 @@ volumes: Execute the following command to start upgrade process: ```bash -docker compose -f docker-compose-upgrade.yml up +docker compose -f docker-compose-upgrade.yml up --abort-on-container-exit {:copy-code} -``` - -Once upgrade process successfully completed, exit from the docker-compose shell by this combination: - -```text -Ctrl + C -``` - -Execute the following command to stop TB Edge upgrade container: - -```bash -docker compose -f docker-compose-upgrade.yml stop -{:copy-code} -``` +``` \ No newline at end of file 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 73a504245d..471c47dafc 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 @@ -404,6 +404,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void scheduleEdgeEventsCheck(EdgeGrpcSession session) { EdgeId edgeId = session.getEdge().getId(); TenantId tenantId = session.getEdge().getTenantId(); + + cancelScheduleEdgeEventsCheck(edgeId); + if (sessions.containsKey(edgeId)) { ScheduledFuture edgeEventCheckTask = edgeEventProcessingExecutorService.schedule(() -> { try { diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 137c4d5bf5..b8cedf8a56 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -448,7 +448,11 @@ public abstract class EdgeGrpcSession implements Closeable { private void scheduleDownlinkMsgsPackSend(int attempt) { Runnable sendDownlinkMsgsTask = () -> { try { - if (isConnected() && !sessionState.getPendingMsgsMap().values().isEmpty()) { + if (!isConnected()) { + stopCurrentSendDownlinkMsgsTask(true); + return; + } + if (!sessionState.getPendingMsgsMap().values().isEmpty()) { List copy = new ArrayList<>(sessionState.getPendingMsgsMap().values()); if (attempt > 1) { String error = "Failed to deliver the batch"; @@ -525,11 +529,11 @@ public abstract class EdgeGrpcSession implements Closeable { private void onDownlinkResponse(DownlinkResponseMsg msg) { try { if (msg.getSuccess()) { - sessionState.getPendingMsgsMap().remove(msg.getDownlinkMsgId()); log.debug("[{}][{}][{}] Msg has been processed successfully! Msg Id: [{}], Msg: {}", tenantId, edge.getId(), sessionId, msg.getDownlinkMsgId(), msg); } else { log.error("[{}][{}][{}] Msg processing failed! Msg Id: [{}], Error msg: {}", tenantId, edge.getId(), sessionId, msg.getDownlinkMsgId(), msg.getErrorMsg()); } + sessionState.getPendingMsgsMap().remove(msg.getDownlinkMsgId()); if (sessionState.getPendingMsgsMap().isEmpty()) { log.debug("[{}][{}][{}] Pending msgs map is empty. Stopping current iteration", tenantId, edge.getId(), sessionId); stopCurrentSendDownlinkMsgsTask(false); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java index faf001fbba..39007552ad 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java @@ -41,7 +41,10 @@ public abstract class BaseDashboardProcessor extends BaseEdgeProcessor { if (dashboard == null) { throw new RuntimeException("[{" + tenantId + "}] dashboardUpdateMsg {" + dashboardUpdateMsg + "} cannot be converted to dashboard"); } - Set newAssignedCustomers = new HashSet<>(dashboard.getAssignedCustomers()); + Set newAssignedCustomers = new HashSet<>(); + if (dashboard.getAssignedCustomers() != null && !dashboard.getAssignedCustomers().isEmpty()) { + newAssignedCustomers.addAll(dashboard.getAssignedCustomers()); + } Dashboard dashboardById = edgeCtx.getDashboardService().findDashboardById(tenantId, dashboardId); if (dashboardById == null) { created = true; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java index 6d47c16680..e04d4d01af 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java @@ -28,17 +28,22 @@ import org.thingsboard.server.gen.edge.v1.WidgetBundleTypesRequestMsg; public interface EdgeRequestsService { + @Deprecated(since = "3.9.1", forRemoval = true) ListenableFuture processRuleChainMetadataRequestMsg(TenantId tenantId, Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg); ListenableFuture processAttributesRequestMsg(TenantId tenantId, Edge edge, AttributesRequestMsg attributesRequestMsg); ListenableFuture processRelationRequestMsg(TenantId tenantId, Edge edge, RelationRequestMsg relationRequestMsg); + @Deprecated(since = "3.9.1", forRemoval = true) ListenableFuture processDeviceCredentialsRequestMsg(TenantId tenantId, Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg); + @Deprecated(since = "3.9.1", forRemoval = true) ListenableFuture processUserCredentialsRequestMsg(TenantId tenantId, Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg); + @Deprecated(since = "3.9.1", forRemoval = true) ListenableFuture processWidgetBundleTypesRequestMsg(TenantId tenantId, Edge edge, WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg); + @Deprecated(since = "3.9.1", forRemoval = true) ListenableFuture processEntityViewsRequestMsg(TenantId tenantId, Edge edge, EntityViewsRequestMsg entityViewsRequestMsg); } diff --git a/application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java index 5b09e05a09..42b75b5a74 100644 --- a/application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java @@ -48,12 +48,14 @@ public class DashboardEdgeTest extends AbstractEdgeTest { private static final int MOBILE_ORDER = 5; private static final String IMAGE = "data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAyNCAyNCIgd2lkdGg9IjQ4IiBoZWlnaHQ9IjQ4Ij48cGF0aCBkPSJNMTMuMjMgMTAuNTZWMTBjLTEuOTQgMC0zLjk5LjM5LTMuOTkgMi42NyAwIDEuMTYuNjEgMS45NSAxLjYzIDEuOTUuNzYgMCAxLjQzLS40NyAxLjg2LTEuMjIuNTItLjkzLjUtMS44LjUtMi44NG0yLjcgNi41M2MtLjE4LjE2LS40My4xNy0uNjMuMDYtLjg5LS43NC0xLjA1LTEuMDgtMS41NC0xLjc5LTEuNDcgMS41LTIuNTEgMS45NS00LjQyIDEuOTUtMi4yNSAwLTQuMDEtMS4zOS00LjAxLTQuMTcgMC0yLjE4IDEuMTctMy42NCAyLjg2LTQuMzggMS40Ni0uNjQgMy40OS0uNzYgNS4wNC0uOTNWNy41YzAtLjY2LjA1LTEuNDEtLjMzLTEuOTYtLjMyLS40OS0uOTUtLjctMS41LS43LTEuMDIgMC0xLjkzLjUzLTIuMTUgMS42MS0uMDUuMjQtLjI1LjQ4LS40Ny40OWwtMi42LS4yOGMtLjIyLS4wNS0uNDYtLjIyLS40LS41Ni42LTMuMTUgMy40NS00LjEgNi00LjEgMS4zIDAgMyAuMzUgNC4wMyAxLjMzQzE3LjExIDQuNTUgMTcgNi4xOCAxNyA3Ljk1djQuMTdjMCAxLjI1LjUgMS44MSAxIDIuNDguMTcuMjUuMjEuNTQgMCAuNzFsLTIuMDYgMS43OGgtLjAxIj48L3BhdGg+PHBhdGggZD0iTTIwLjE2IDE5LjU0QzE4IDIxLjE0IDE0LjgyIDIyIDEyLjEgMjJjLTMuODEgMC03LjI1LTEuNDEtOS44NS0zLjc2LS4yLS4xOC0uMDItLjQzLjI1LS4yOSAyLjc4IDEuNjMgNi4yNSAyLjYxIDkuODMgMi42MSAyLjQxIDAgNS4wNy0uNSA3LjUxLTEuNTMuMzctLjE2LjY2LjI0LjMyLjUxIj48L3BhdGg+PHBhdGggZD0iTTIxLjA3IDE4LjVjLS4yOC0uMzYtMS44NS0uMTctMi41Ny0uMDgtLjE5LjAyLS4yMi0uMTYtLjAzLS4zIDEuMjQtLjg4IDMuMjktLjYyIDMuNTMtLjMzLjI0LjMtLjA3IDIuMzUtMS4yNCAzLjMyLS4xOC4xNi0uMzUuMDctLjI2LS4xMS4yNi0uNjcuODUtMi4xNC41Ny0yLjV6Ij48L3BhdGg+PC9zdmc+"; + private static final String DASHBOARD_TITLE = "Edge Test Dashboard"; + @Test public void testDashboards() throws Exception { // create dashboard and assign to edge edgeImitator.expectMessageAmount(2); Dashboard dashboard = new Dashboard(); - dashboard.setTitle("Edge Test Dashboard"); + dashboard.setTitle(DASHBOARD_TITLE); dashboard.setMobileHide(true); dashboard.setImage(IMAGE); dashboard.setMobileOrder(MOBILE_ORDER); @@ -200,12 +202,27 @@ public class DashboardEdgeTest extends AbstractEdgeTest { Dashboard foundDashboard = doGet("/api/dashboard/" + dashboard.getUuidId(), Dashboard.class); Assert.assertNotNull(foundDashboard); - Assert.assertEquals("Edge Test Dashboard", foundDashboard.getName()); + Assert.assertEquals(DASHBOARD_TITLE, foundDashboard.getName()); PageData pageData = doGetTypedWithPageLink("/api/customer/" + savedCustomer.getId().toString() + "/dashboards?", new TypeReference<>() {}, new PageLink(100)); Assert.assertEquals(1, pageData.getData().size()); - Assert.assertEquals("Edge Test Dashboard", pageData.getData().get(0).getTitle()); + Assert.assertEquals(DASHBOARD_TITLE, pageData.getData().get(0).getTitle()); + + dashboard.setTitle(DASHBOARD_TITLE + " Updated"); + dashboard.setAssignedCustomers(null); + dashboardUpdateMsgBuilder.setEntity(JacksonUtil.toString(dashboard)); + dashboardUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE); + uplinkMsgBuilder = UplinkMsg.newBuilder(); + uplinkMsgBuilder.addDashboardUpdateMsg(dashboardUpdateMsgBuilder.build()); + + edgeImitator.expectResponsesAmount(1); + edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build()); + + Assert.assertTrue(edgeImitator.waitForResponses()); + + foundDashboard = doGet("/api/dashboard/" + dashboard.getUuidId(), Dashboard.class); + Assert.assertEquals(DASHBOARD_TITLE + " Updated", foundDashboard.getName()); } @Test @@ -256,7 +273,7 @@ public class DashboardEdgeTest extends AbstractEdgeTest { Dashboard dashboard = new Dashboard(); dashboard.setId(new DashboardId(UUID.randomUUID())); dashboard.setTenantId(tenantId); - dashboard.setTitle("Edge Test Dashboard"); + dashboard.setTitle(DASHBOARD_TITLE); dashboard.setAssignedCustomers(Sets.newHashSet(new ShortCustomerInfo(savedCustomer.getId(), savedCustomer.getTitle(), savedCustomer.isPublic()))); return dashboard; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index dd5d61e834..cd9d577b86 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -472,7 +472,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(topicService.buildTopicName("tb_edge_event.notifications." + tenantId + "." + edgeId)); - consumerBuilder.clientId("monolith-to-edge-event-consumer" + serviceInfoProvider.getServiceId()); + consumerBuilder.clientId("monolith-to-edge-event-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); consumerBuilder.groupId(topicService.buildTopicName("monolith-edge-event-consumer")); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToEdgeEventNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(edgeEventAdmin); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index cc0e044917..dffb127c70 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -421,7 +421,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(topicService.buildTopicName("tb_edge_event.notifications." + tenantId + "." + edgeId)); - consumerBuilder.clientId("tb-core-edge-event-consumer" + serviceInfoProvider.getServiceId()); + consumerBuilder.clientId("tb-core-edge-event-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); consumerBuilder.groupId(topicService.buildTopicName("tb-core-edge-event-consumer")); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToEdgeEventNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(edgeEventAdmin);