Merge branch 'fix/teambition/2024-10-31' into 'master_dev'

perf:优化边端边接日志打印 See merge request yunteng/thingskit!465

Merge branch 'fix/teambition/2024-10-31' into 'master_dev'
perf:优化边端边接日志打印 See merge request yunteng/thingskit!465
xp.Huang
2 parents ae98cb4d 4b64b5f6
Showing 1 changed file with 21 additions and 14 deletions
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View file @9242811
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View file @9242811
@@ -131,20 +131,20 @@ public final class EdgeGrpcSession implements Closeable {
             public void onNext(RequestMsg requestMsg) {
                 if (!connected && requestMsg.getMsgType().equals(RequestMsgType.CONNECT_RPC_MESSAGE)) {
                     ConnectResponseMsg responseMsg = processConnect(requestMsg.getConnectRequestMsg());
-                    lifecycleEvent(getTenantId(), getEdge().getId(), ComponentLifecycleEvent.CREATED, true, null);
+                    lifecycleEvent(getTenantId(), getEdge(), ComponentLifecycleEvent.CREATED, true, null);
                     outputStream.onNext(ResponseMsg.newBuilder()
                             .setConnectResponseMsg(responseMsg)
                             .build());
                     if (ConnectResponseCode.ACCEPTED != responseMsg.getResponseCode()) {
                         outputStream.onError(new RuntimeException(responseMsg.getErrorMsg()));
-                        lifecycleEvent(getTenantId(), getEdge().getId(), ComponentLifecycleEvent.FAILED, false, new RuntimeException(responseMsg.getErrorMsg()));
+                        lifecycleEvent(getTenantId(), getEdge(), ComponentLifecycleEvent.FAILED, false, new RuntimeException(responseMsg.getErrorMsg()));
                     } else {
                         if (requestMsg.getConnectRequestMsg().hasMaxInboundMessageSize()) {
                             log.debug("[{}][{}] Client max inbound message size: {}", tenantId, sessionId, requestMsg.getConnectRequestMsg().getMaxInboundMessageSize());
                             clientMaxInboundMessageSize = requestMsg.getConnectRequestMsg().getMaxInboundMessageSize();
                         }
                         connected = true;
-                        lifecycleEvent(getTenantId(), getEdge().getId(), ComponentLifecycleEvent.STARTED, true, null);
+                        lifecycleEvent(getTenantId(), getEdge(), ComponentLifecycleEvent.STARTED, true, null);
                     }
                 }
                 if (connected) {
@@ -174,7 +174,7 @@ public final class EdgeGrpcSession implements Closeable {
             @Override
             public void onError(Throwable t) {
                 log.error("[{}][{}] Stream was terminated due to error:", tenantId, sessionId, t);
-                lifecycleEvent(getTenantId(), getEdge().getId(), ComponentLifecycleEvent.FAILED, false, t);
+                lifecycleEvent(getTenantId(), getEdge(), ComponentLifecycleEvent.FAILED, false, t);
                 closeSession();
             }
@@ -189,15 +189,15 @@ public final class EdgeGrpcSession implements Closeable {
                 if (edge != null) {
                     try {
                         sessionCloseListener.accept(edge, sessionId);
-                        lifecycleEvent(getTenantId(), getEdge().getId(), ComponentLifecycleEvent.STOPPED, true, null);
+                        lifecycleEvent(getTenantId(), getEdge(), ComponentLifecycleEvent.STOPPED, true, null);
                     } catch (Exception ignored) {
-                        lifecycleEvent(getTenantId(), getEdge().getId(), ComponentLifecycleEvent.FAILED, false, ignored);
+                        lifecycleEvent(getTenantId(), getEdge(), ComponentLifecycleEvent.FAILED, false, ignored);
                     }
                 }
                 try {
                     outputStream.onCompleted();
                 } catch (Exception ignored) {
-                    lifecycleEvent(getTenantId(), getEdge().getId(), ComponentLifecycleEvent.FAILED, false, ignored);
+                    lifecycleEvent(getTenantId(), getEdge(), ComponentLifecycleEvent.FAILED, false, ignored);
                 }
             }
         };
@@ -208,7 +208,7 @@ public final class EdgeGrpcSession implements Closeable {
         syncCompleted = false;
         interruptGeneralProcessingOnSync();
         doSync(new EdgeSyncCursor(ctx, edge, fullSync));
-        lifecycleEvent(getTenantId(), getEdge().getId(), ComponentLifecycleEvent.ACTIVATED, true, null);
+        lifecycleEvent(getTenantId(), getEdge(), ComponentLifecycleEvent.ACTIVATED, true, null);
     }
     private void doSync(EdgeSyncCursor cursor) {
@@ -333,7 +333,7 @@ public final class EdgeGrpcSession implements Closeable {
                 .setEdgeUpdateMsg(edgeConfig)
                 .build();
         sendDownlinkMsg(edgeConfigMsg);
-        lifecycleEvent(getTenantId(), getEdge().getId(), ComponentLifecycleEvent.UPDATED, true, null);
+        lifecycleEvent(getTenantId(), getEdge(), ComponentLifecycleEvent.UPDATED, true, null);
     }
     ListenableFuture<Boolean> processEdgeEvents() throws Exception {
@@ -873,6 +873,8 @@ public final class EdgeGrpcSession implements Closeable {
                         .setErrorMsg(failureMsg)
                         .setConfiguration(EdgeConfiguration.getDefaultInstance()).build();
             }
+        } else {
+            log.warn("非法边端连接,EdgeRoutingKey [{}] 不存在",request.getEdgeRoutingKey());
         }
         return ConnectResponseMsg.newBuilder()
                 .setResponseCode(ConnectResponseCode.BAD_CREDENTIALS)
@@ -918,23 +920,28 @@ public final class EdgeGrpcSession implements Closeable {
         }
     }
 	//thingskit
-    private void lifecycleEvent(TenantId tenantId, EdgeId edgeId, ComponentLifecycleEvent eventType, boolean success, Throwable error) {
+    private void lifecycleEvent(TenantId tenantId, Edge edge, ComponentLifecycleEvent eventType, boolean success, Throwable error) {
         try {
+
+            if(tenantId==null || edge==null || edge.getId()==null){
+                return;
+            }
+
             ToCoreMsg msg = ToCoreMsg.newBuilder()
                     .setLifecycleEventMsg(
                             LifecycleEventProto.newBuilder()
                                     .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
                                     .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
-                                    .setEntityIdMSB(edgeId.getId().getMostSignificantBits())
-                                    .setEntityIdLSB(edgeId.getId().getLeastSignificantBits())
+                                    .setEntityIdMSB(edge.getId().getId().getMostSignificantBits())
+                                    .setEntityIdLSB(edge.getId().getId().getLeastSignificantBits())
                                     .setServiceId(ctx.getServiceInfoProvider().getServiceId())
                                     .setLcEventType(eventType.name())
                                     .setSuccess(success)
                                     .setError(error != null ? ExceptionUtils.getStackTrace(error) : "")
                     ).build();
-            ctx.getClusterService().pushMsgToCore(tenantId, (EntityId) edgeId, msg, null);
+            ctx.getClusterService().pushMsgToCore(tenantId, (EntityId) edge.getId(), msg, null);
         } catch (Exception e) {
-            log.error("[{}][{}] Failed to send lifecycle event to core", tenantId, edgeId, e);
+            log.error("[{}][{}] Failed to send lifecycle event to core", tenantId, edge.getId(), e);
         }
     }