Merge pull request #10151 from AndriiLandiak/fix/edge-support

Edge - minor improvement
thingsboard · Feb 13, 2024 · 9f780a7 · 9f780a7
2 parents 2fb9b74 + a4c5617
commit 9f780a7
Show file tree

Hide file tree

Showing 17 changed files with 173 additions and 71 deletions.
diff --git a/...ation/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/...ation/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
@@ -66,7 +66,6 @@
 import org.thingsboard.server.common.msg.rule.engine.DeviceEdgeUpdateMsg;
 import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg;
 import org.thingsboard.server.common.msg.timeout.DeviceActorServerSideRpcTimeoutMsg;
-import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.DeviceSessionsCacheEntry;
@@ -90,7 +89,9 @@
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportUpdateCredentialsProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
+import org.thingsboard.server.gen.transport.TransportProtos.UplinkNotificationMsg;
 import org.thingsboard.server.service.rpc.RpcSubmitStrategy;
+import org.thingsboard.server.service.state.DefaultDeviceStateService;
 import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
 
 import javax.annotation.Nullable;
@@ -173,7 +174,7 @@ boolean initAttributes() {
 
     private EdgeId findRelatedEdgeId() {
         List<EntityRelation> result =
-                systemContext.getRelationService().findByToAndType(tenantId, deviceId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.COMMON);
+                systemContext.getRelationService().findByToAndType(tenantId, deviceId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE);
         if (result != null && result.size() > 0) {
             EntityRelation relationToEdge = result.get(0);
             if (relationToEdge.getFrom() != null && relationToEdge.getFrom().getId() != null) {
@@ -212,8 +213,11 @@ void processRpcRequest(TbActorCtx context, ToDeviceRpcRequestActorMsg msg) {
         if (systemContext.isEdgesEnabled() && edgeId != null) {
             log.debug("[{}][{}] device is related to edge: [{}]. Saving RPC request: [{}][{}] to edge queue", tenantId, deviceId, edgeId.getId(), rpcId, requestId);
             try {
-                saveRpcRequestToEdgeQueue(request, requestId).get();
-                sent = true;
+                if (systemContext.getEdgeService().isEdgeActiveAsync(tenantId, edgeId, DefaultDeviceStateService.ACTIVITY_STATE).get()) {
+                    saveRpcRequestToEdgeQueue(request, requestId).get();
+                } else {
+                    log.error("[{}][{}][{}] Failed to save RPC request to edge queue {}. The Edge is currently offline or unreachable", tenantId, deviceId, edgeId.getId(), request);
+                }
             } catch (InterruptedException | ExecutionException e) {
                 log.error("[{}][{}][{}] Failed to save RPC request to edge queue {}", tenantId, deviceId, edgeId.getId(), request, e);
             }
@@ -470,7 +474,7 @@ void process(TransportToDeviceActorMsgWrapper wrapper) {
         callback.onSuccess();
     }
 
-    private void processUplinkNotificationMsg(SessionInfoProto sessionInfo, TransportProtos.UplinkNotificationMsg uplinkNotificationMsg) {
+    private void processUplinkNotificationMsg(SessionInfoProto sessionInfo, UplinkNotificationMsg uplinkNotificationMsg) {
         String nodeId = sessionInfo.getNodeId();
         sessions.entrySet().stream()
                 .filter(kv -> kv.getValue().getSessionInfo().getNodeId().equals(nodeId) && (kv.getValue().isSubscribedToAttributes() || kv.getValue().isSubscribedToRPC()))

diff --git a/...src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/...src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
@@ -16,7 +16,6 @@
 package org.thingsboard.server.actors.ruleChain;
 
 import lombok.extern.slf4j.Slf4j;
-import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.TbActorCtx;
 import org.thingsboard.server.actors.TbActorRef;
@@ -29,6 +28,7 @@
 import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
 import org.thingsboard.server.common.data.plugin.ComponentLifecycleState;
 import org.thingsboard.server.common.data.relation.EntityRelation;

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
@@ -477,6 +477,8 @@ private void pushRuleEngineMessage(TenantId tenantId, Edge edge, long ts, TbMsgT
             TbMsgMetaData md = new TbMsgMetaData();
             if (!persistToTelemetry) {
                 md.putValue(DataConstants.SCOPE, DataConstants.SERVER_SCOPE);
+                md.putValue("edgeName", edge.getName());
+                md.putValue("edgeType", edge.getType());
             }
             TbMsg tbMsg = TbMsg.newMsg(msgType, edgeId, md, TbMsgDataType.JSON, data);
             clusterService.pushMsgToRuleEngine(tenantId, edgeId, tbMsg, null);

diff --git a/...n/java/org/thingsboard/server/service/edge/rpc/processor/rule/RuleChainEdgeProcessor.java b/...n/java/org/thingsboard/server/service/edge/rpc/processor/rule/RuleChainEdgeProcessor.java
@@ -18,6 +18,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.thingsboard.server.common.data.EdgeUtils;
+import org.thingsboard.server.common.data.edge.Edge;
 import org.thingsboard.server.common.data.edge.EdgeEvent;
 import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.rule.RuleChain;
@@ -53,6 +54,10 @@ public DownlinkMsg convertRuleChainEventToDownlink(EdgeEvent edgeEvent, EdgeVers
                             isRoot = Boolean.parseBoolean(edgeEvent.getBody().get(EDGE_IS_ROOT_BODY_KEY).asText());
                         } catch (Exception ignored) {}
                     }
+                    if (!isRoot) {
+                        Edge edge = edgeService.findEdgeById(edgeEvent.getTenantId(), edgeEvent.getEdgeId());
+                        isRoot = edge.getRootRuleChainId().equals(ruleChainId);
+                    }
                     UpdateMsgType msgType = getUpdateMsgType(edgeEvent.getAction());
                     RuleChainUpdateMsg ruleChainUpdateMsg = ((RuleChainMsgConstructor)
                             ruleChainMsgConstructorFactory.getMsgConstructorByEdgeVersion(edgeVersion))

diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java
@@ -24,7 +24,6 @@
 import org.thingsboard.server.common.msg.queue.RuleEngineException;
 import org.thingsboard.server.common.msg.queue.RuleNodeInfo;
 import org.thingsboard.server.common.msg.queue.TbMsgCallback;
-import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
 
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;

diff --git a/application/src/test/java/org/thingsboard/server/edge/DeviceEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/DeviceEdgeTest.java
@@ -711,7 +711,7 @@ public void sendUpdateSharedAttributeToCloudAndValidateDeviceSubscription() thro
         edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
         Assert.assertTrue(edgeImitator.waitForResponses());
 
-        Assert.assertTrue(onUpdateCallback.getSubscribeLatch().await(30, TimeUnit.SECONDS));
+        Assert.assertTrue(onUpdateCallback.getSubscribeLatch().await(TIMEOUT, TimeUnit.SECONDS));
 
         Assert.assertEquals(JacksonUtil.newObjectNode().put(attrKey, attrValue),
                 JacksonUtil.fromBytes(onUpdateCallback.getPayloadBytes()));
@@ -798,7 +798,21 @@ public void testVerifyProcessCorrectEdgeUpdateToDeviceActorOnUnassignFromDiffere
         // clean up stored edge events
         edgeEventService.cleanupEvents(1);
 
-        // perform rpc call to verify edgeId in DeviceActorMessageProcessor updated properly
+        // edge is disconnected: perform rpc call - no edge event saved
+        doPostAsync(
+                "/api/rpc/oneway/" + device.getId().getId().toString(),
+                JacksonUtil.toString(createDefaultRpc()),
+                String.class,
+                status().isOk());
+        Awaitility.await()
+                .atMost(TIMEOUT, TimeUnit.SECONDS)
+                .until(() -> {
+                    PageData<EdgeEvent> result = edgeEventService.findEdgeEvents(tenantId, tmpEdge.getId(), 0L, null, new TimePageLink(1));
+                    return result.getTotalElements() == 0;
+                });
+
+        // edge is connected: perform rpc call to verify edgeId in DeviceActorMessageProcessor updated properly
+        simulateEdgeActivation(tmpEdge);
         doPostAsync(
                 "/api/rpc/oneway/" + device.getId().getId().toString(),
                 JacksonUtil.toString(createDefaultRpc()),
@@ -857,4 +871,23 @@ private ObjectNode createDefaultRpc() {
 
         return rpc;
     }
+
+    private void simulateEdgeActivation(Edge edge) throws Exception {
+        ObjectNode attributes = JacksonUtil.newObjectNode();
+        attributes.put("active", true);
+        doPost("/api/plugins/telemetry/EDGE/" + edge.getId() + "/attributes/" + DataConstants.SERVER_SCOPE, attributes);
+        Awaitility.await()
+                .atMost(TIMEOUT, TimeUnit.SECONDS)
+                .until(() -> {
+                    List<Map<String, Object>> values = doGetAsyncTyped("/api/plugins/telemetry/EDGE/" + edge.getId() +
+                            "/values/attributes/SERVER_SCOPE", new TypeReference<>() {});
+                    Optional<Map<String, Object>> activeAttrOpt = values.stream().filter(att -> att.get("key").equals("active")).findFirst();
+                    if (activeAttrOpt.isEmpty()) {
+                        return false;
+                    }
+                    Map<String, Object> activeAttr = activeAttrOpt.get();
+                    return "true".equals(activeAttr.get("value").toString());
+                });
+    }
+
 }
diff --git a/application/src/test/java/org/thingsboard/server/edge/RuleChainEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/RuleChainEdgeTest.java
@@ -146,7 +146,7 @@ private void testRuleChainMetadataRequestMsg(RuleChainId ruleChainId) throws Exc
         }
     }
 
-    private void createRuleChainMetadata(RuleChain ruleChain) {
+    private RuleChainMetaData createRuleChainMetadata(RuleChain ruleChain) {
         RuleChainMetaData ruleChainMetaData = new RuleChainMetaData();
         ruleChainMetaData.setRuleChainId(ruleChain.getId());
 
@@ -182,7 +182,7 @@ private void createRuleChainMetadata(RuleChain ruleChain) {
         ruleChainMetaData.addConnectionInfo(0, 2, "fail");
         ruleChainMetaData.addConnectionInfo(1, 2, "success");
 
-        doPost("/api/ruleChain/metadata", ruleChainMetaData, RuleChainMetaData.class);
+        return doPost("/api/ruleChain/metadata", ruleChainMetaData, RuleChainMetaData.class);
     }
 
     @Test
@@ -193,9 +193,10 @@ public void testSetRootRuleChain() throws Exception {
         ruleChain.setType(RuleChainType.EDGE);
         RuleChain savedRuleChain = doPost("/api/ruleChain", ruleChain, RuleChain.class);
 
-        edgeImitator.expectMessageAmount(1);
+        edgeImitator.expectMessageAmount(2);
         doPost("/api/edge/" + edge.getUuidId()
                 + "/ruleChain/" + savedRuleChain.getUuidId(), RuleChain.class);
+        RuleChainMetaData metaData = createRuleChainMetadata(savedRuleChain);
         Assert.assertTrue(edgeImitator.waitForMessages());
 
         // set new rule chain as root
@@ -213,6 +214,22 @@ public void testSetRootRuleChain() throws Exception {
         Assert.assertTrue(ruleChainMsg.isRoot());
         Assert.assertEquals(savedRuleChain.getId(), ruleChainMsg.getId());
 
+        // update metadata for root rule chain
+        edgeImitator.expectMessageAmount(1);
+        metaData.getNodes().forEach(n -> n.setDebugMode(true));
+        doPost("/api/ruleChain/metadata", metaData, RuleChainMetaData.class);
+        Assert.assertTrue(edgeImitator.waitForMessages());
+        ruleChainUpdateMsgOpt = edgeImitator.findMessageByType(RuleChainUpdateMsg.class);
+        Assert.assertTrue(ruleChainUpdateMsgOpt.isPresent());
+        ruleChainUpdateMsg = ruleChainUpdateMsgOpt.get();
+        ruleChainMsg = JacksonUtil.fromString(ruleChainUpdateMsg.getEntity(), RuleChain.class, true);
+        Assert.assertNotNull(ruleChainMsg);
+        Assert.assertTrue(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE.equals(ruleChainUpdateMsg.getMsgType()) ||
+                UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE.equals(ruleChainUpdateMsg.getMsgType()));
+        Assert.assertEquals(savedRuleChain.getId(), ruleChainMsg.getId());
+        Assert.assertEquals(savedRuleChain.getName(), ruleChainMsg.getName());
+        Assert.assertTrue(ruleChainMsg.isRoot());
+
         // revert root rule chain
         edgeImitator.expectMessageAmount(1);
         doPost("/api/edge/" + edge.getUuidId()

diff --git a/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java b/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java
@@ -26,6 +26,7 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.thingsboard.edge.rpc.EdgeGrpcClient;
 import org.thingsboard.edge.rpc.EdgeRpcClient;
+import org.thingsboard.server.controller.AbstractWebTest;
 import org.thingsboard.server.gen.edge.v1.AdminSettingsUpdateMsg;
 import org.thingsboard.server.gen.edge.v1.AlarmCommentUpdateMsg;
 import org.thingsboard.server.gen.edge.v1.AlarmUpdateMsg;
@@ -72,8 +73,6 @@
 @Slf4j
 public class EdgeImitator {
 
-    public static final int TIMEOUT_IN_SECONDS = 30;
-
     private String routingKey;
     private String routingSecret;
 
@@ -344,7 +343,7 @@ private ListenableFuture<Void> saveDownlinkMsg(AbstractMessage message) {
     }
 
     public boolean waitForMessages() throws InterruptedException {
-        return waitForMessages(TIMEOUT_IN_SECONDS);
+        return waitForMessages(AbstractWebTest.TIMEOUT);
     }
 
     public boolean waitForMessages(int timeoutInSeconds) throws InterruptedException {
@@ -359,7 +358,7 @@ public void expectMessageAmount(int messageAmount) {
     }
 
     public boolean waitForResponses() throws InterruptedException {
-        return responsesLatch.await(TIMEOUT_IN_SECONDS, TimeUnit.SECONDS);
+        return responsesLatch.await(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS);
     }
 
     public void expectResponsesAmount(int messageAmount) {