dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improved Edge handling in case connect/disconnect. Speed up initial setup

Browse Source
This commit is contained in:
Volodymyr Babak 2025-02-10 18:42:25 +02:00
parent 3eaa885f89
commit 3860d79613
8 changed files with 43 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md
View File

@ -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}
```
```

3
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
View File

@ -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 {

8
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View File

@ -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<DownlinkMsg> 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);

5
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java
View File

@ -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<ShortCustomerInfo> newAssignedCustomers = new HashSet<>(dashboard.getAssignedCustomers());
Set<ShortCustomerInfo> 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;

5
application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java
View File

@ -28,17 +28,22 @@ import org.thingsboard.server.gen.edge.v1.WidgetBundleTypesRequestMsg;
public interface EdgeRequestsService {
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processRuleChainMetadataRequestMsg(TenantId tenantId, Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg);
ListenableFuture<Void> processAttributesRequestMsg(TenantId tenantId, Edge edge, AttributesRequestMsg attributesRequestMsg);
ListenableFuture<Void> processRelationRequestMsg(TenantId tenantId, Edge edge, RelationRequestMsg relationRequestMsg);
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processDeviceCredentialsRequestMsg(TenantId tenantId, Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg);
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processUserCredentialsRequestMsg(TenantId tenantId, Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg);
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processWidgetBundleTypesRequestMsg(TenantId tenantId, Edge edge, WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg);
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processEntityViewsRequestMsg(TenantId tenantId, Edge edge, EntityViewsRequestMsg entityViewsRequestMsg);
}

25
application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java
View File

@ -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<DashboardInfo> 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;
}

2
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java
View File

@ -472,7 +472,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> 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);

2
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java
View File

@ -421,7 +421,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> 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);