From 0b9f7f0518ae30e1e6551fea3bac13b845f74096 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Fri, 1 Sep 2023 16:53:52 +0300 Subject: [PATCH] Added logs for newly added functionality. Added tenantId into edge logs --- .../edge/DefaultEdgeNotificationService.java | 14 +-- .../service/edge/rpc/EdgeGrpcService.java | 50 ++++++----- .../service/edge/rpc/EdgeGrpcSession.java | 86 ++++++++++--------- .../rpc/constructor/AlarmMsgConstructor.java | 43 +--------- .../constructor/EntityDataMsgConstructor.java | 11 +-- .../AbstractRuleChainMetadataConstructor.java | 2 +- .../fetch/AdminSettingsEdgeEventFetcher.java | 6 +- .../processor/alarm/BaseAlarmProcessor.java | 37 +++++++- .../asset/AssetProfileEdgeProcessor.java | 8 +- .../processor/asset/BaseAssetProcessor.java | 7 +- .../asset/BaseAssetProfileProcessor.java | 7 +- .../processor/device/BaseDeviceProcessor.java | 19 ++-- .../device/BaseDeviceProfileProcessor.java | 9 +- .../device/DeviceProfileEdgeProcessor.java | 9 +- .../rpc/processor/edge/EdgeProcessor.java | 4 +- .../entityview/BaseEntityViewProcessor.java | 4 +- .../relation/BaseRelationProcessor.java | 2 +- .../telemetry/BaseTelemetryProcessor.java | 25 +++--- .../telemetry/TelemetryEdgeProcessor.java | 3 +- .../rpc/sync/DefaultEdgeRequestsService.java | 14 +-- .../server/edge/AssetProfileEdgeTest.java | 54 +++++++++++- .../server/edge/DashboardEdgeTest.java | 10 ++- .../server/edge/DeviceProfileEdgeTest.java | 47 +++++++++- .../server/edge/WidgetEdgeTest.java | 2 + 24 files changed, 291 insertions(+), 182 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java index e19cb80d4c..69852e6c05 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java @@ -155,9 +155,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @Override public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) { - log.debug("Pushing notification to edge {}", edgeNotificationMsg); + TenantId tenantId = TenantId.fromUUID(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); + log.debug("[{}] Pushing notification to edge {}", tenantId, edgeNotificationMsg); try { - TenantId tenantId = TenantId.fromUUID(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); ListenableFuture future; switch (type) { @@ -216,7 +216,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { future = tenantProfileEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break; default: - log.warn("Edge event type [{}] is not designed to be pushed to edge", type); + log.warn("[{}] Edge event type [{}] is not designed to be pushed to edge", tenantId, type); future = Futures.immediateFuture(null); } Futures.addCallback(future, new FutureCallback<>() { @@ -227,16 +227,16 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @Override public void onFailure(Throwable throwable) { - callBackFailure(edgeNotificationMsg, callback, throwable); + callBackFailure(tenantId, edgeNotificationMsg, callback, throwable); } }, dbCallBackExecutor); } catch (Exception e) { - callBackFailure(edgeNotificationMsg, callback, e); + callBackFailure(tenantId, edgeNotificationMsg, callback, e); } } - private void callBackFailure(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback, Throwable throwable) { - log.error("Can't push to edge updates, edgeNotificationMsg [{}]", edgeNotificationMsg, throwable); + private void callBackFailure(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback, Throwable throwable) { + log.error("[{}] Can't push to edge updates, edgeNotificationMsg [{}]", tenantId, edgeNotificationMsg, throwable); callback.onFailure(throwable); } 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 d002854155..50b80a82f7 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 @@ -195,17 +195,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i switch (msg.getMsgType()) { case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG: EdgeEventUpdateMsg edgeEventUpdateMsg = (EdgeEventUpdateMsg) msg; - log.trace("[{}] onToEdgeSessionMsg [{}]", edgeEventUpdateMsg.getTenantId(), msg); + log.trace("[{}] onToEdgeSessionMsg [{}]", tenantId, msg); onEdgeEvent(tenantId, edgeEventUpdateMsg.getEdgeId()); break; case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG: ToEdgeSyncRequest toEdgeSyncRequest = (ToEdgeSyncRequest) msg; - log.trace("[{}] toEdgeSyncRequest [{}]", toEdgeSyncRequest.getTenantId(), msg); + log.trace("[{}] toEdgeSyncRequest [{}]", tenantId, msg); startSyncProcess(tenantId, toEdgeSyncRequest.getEdgeId(), toEdgeSyncRequest.getId()); break; case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG: FromEdgeSyncResponse fromEdgeSyncResponse = (FromEdgeSyncResponse) msg; - log.trace("[{}] fromEdgeSyncResponse [{}]", fromEdgeSyncResponse.getTenantId(), msg); + log.trace("[{}] fromEdgeSyncResponse [{}]", tenantId, msg); processSyncResponse(fromEdgeSyncResponse); break; } @@ -263,7 +263,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } private void onEdgeConnect(EdgeId edgeId, EdgeGrpcSession edgeGrpcSession) { - log.info("[{}] edge [{}] connected successfully.", edgeGrpcSession.getSessionId(), edgeId); + TenantId tenantId = edgeGrpcSession.getEdge().getTenantId(); + log.info("[{}][{}] edge [{}] connected successfully.", tenantId, edgeGrpcSession.getSessionId(), edgeId); sessions.put(edgeId, edgeGrpcSession); final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); newEventLock.lock(); @@ -272,10 +273,10 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } finally { newEventLock.unlock(); } - save(edgeId, DefaultDeviceStateService.ACTIVITY_STATE, true); + save(tenantId, edgeId, DefaultDeviceStateService.ACTIVITY_STATE, true); long lastConnectTs = System.currentTimeMillis(); - save(edgeId, DefaultDeviceStateService.LAST_CONNECT_TIME, lastConnectTs); - pushRuleEngineMessage(edgeGrpcSession.getEdge().getTenantId(), edgeId, lastConnectTs, TbMsgType.CONNECT_EVENT); + save(tenantId, edgeId, DefaultDeviceStateService.LAST_CONNECT_TIME, lastConnectTs); + pushRuleEngineMessage(tenantId, edgeId, lastConnectTs, TbMsgType.CONNECT_EVENT); cancelScheduleEdgeEventsCheck(edgeId); scheduleEdgeEventsCheck(edgeGrpcSession); } @@ -334,7 +335,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i newEventLock.lock(); try { if (Boolean.TRUE.equals(sessionNewEvents.get(edgeId))) { - log.trace("[{}] Set session new events flag to false", edgeId.getId()); + log.trace("[{}][{}] Set session new events flag to false", tenantId, edgeId.getId()); sessionNewEvents.put(edgeId, false); Futures.addCallback(session.processEdgeEvents(), new FutureCallback<>() { @Override @@ -392,9 +393,10 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } finally { newEventLock.unlock(); } - save(edgeId, DefaultDeviceStateService.ACTIVITY_STATE, false); + TenantId tenantId = toRemove.getEdge().getTenantId(); + save(tenantId, edgeId, DefaultDeviceStateService.ACTIVITY_STATE, false); long lastDisconnectTs = System.currentTimeMillis(); - save(edgeId, DefaultDeviceStateService.LAST_DISCONNECT_TIME, lastDisconnectTs); + save(tenantId, edgeId, DefaultDeviceStateService.LAST_DISCONNECT_TIME, lastDisconnectTs); pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edgeId, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); cancelScheduleEdgeEventsCheck(edgeId); } else { @@ -402,36 +404,38 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } - private void save(EdgeId edgeId, String key, long value) { - log.debug("[{}] Updating long edge telemetry [{}] [{}]", edgeId, key, value); + private void save(TenantId tenantId, EdgeId edgeId, String key, long value) { + log.debug("[{}][{}] Updating long edge telemetry [{}] [{}]", tenantId, edgeId, key, value); if (persistToTelemetry) { tsSubService.saveAndNotify( - TenantId.SYS_TENANT_ID, edgeId, + tenantId, edgeId, Collections.singletonList(new BasicTsKvEntry(System.currentTimeMillis(), new LongDataEntry(key, value))), - new AttributeSaveCallback(edgeId, key, value)); + new AttributeSaveCallback(tenantId, edgeId, key, value)); } else { - tsSubService.saveAttrAndNotify(TenantId.SYS_TENANT_ID, edgeId, DataConstants.SERVER_SCOPE, key, value, new AttributeSaveCallback(edgeId, key, value)); + tsSubService.saveAttrAndNotify(tenantId, edgeId, DataConstants.SERVER_SCOPE, key, value, new AttributeSaveCallback(tenantId, edgeId, key, value)); } } - private void save(EdgeId edgeId, String key, boolean value) { - log.debug("[{}] Updating boolean edge telemetry [{}] [{}]", edgeId, key, value); + private void save(TenantId tenantId, EdgeId edgeId, String key, boolean value) { + log.debug("[{}][{}] Updating boolean edge telemetry [{}] [{}]", tenantId, edgeId, key, value); if (persistToTelemetry) { tsSubService.saveAndNotify( - TenantId.SYS_TENANT_ID, edgeId, + tenantId, edgeId, Collections.singletonList(new BasicTsKvEntry(System.currentTimeMillis(), new BooleanDataEntry(key, value))), - new AttributeSaveCallback(edgeId, key, value)); + new AttributeSaveCallback(tenantId, edgeId, key, value)); } else { - tsSubService.saveAttrAndNotify(TenantId.SYS_TENANT_ID, edgeId, DataConstants.SERVER_SCOPE, key, value, new AttributeSaveCallback(edgeId, key, value)); + tsSubService.saveAttrAndNotify(tenantId, edgeId, DataConstants.SERVER_SCOPE, key, value, new AttributeSaveCallback(tenantId, edgeId, key, value)); } } private static class AttributeSaveCallback implements FutureCallback { + private final TenantId tenantId; private final EdgeId edgeId; private final String key; private final Object value; - AttributeSaveCallback(EdgeId edgeId, String key, Object value) { + AttributeSaveCallback(TenantId tenantId, EdgeId edgeId, String key, Object value) { + this.tenantId = tenantId; this.edgeId = edgeId; this.key = key; this.value = value; @@ -439,12 +443,12 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @Override public void onSuccess(@Nullable Void result) { - log.trace("[{}] Successfully updated attribute [{}] with value [{}]", edgeId, key, value); + log.trace("[{}][{}] Successfully updated attribute [{}] with value [{}]", tenantId, edgeId, key, value); } @Override public void onFailure(Throwable t) { - log.warn("[{}] Failed to update attribute [{}] with value [{}]", edgeId, key, value, t); + log.warn("[{}][{}] Failed to update attribute [{}] with value [{}]", tenantId, edgeId, key, value, t); } } 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 d5d73d47e7..37b71ce494 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 @@ -105,6 +105,7 @@ public final class EdgeGrpcSession implements Closeable { private EdgeContextComponent ctx; private Edge edge; + private TenantId tenantId; private StreamObserver inputStream; private StreamObserver outputStream; private boolean connected; @@ -148,7 +149,7 @@ public final class EdgeGrpcSession implements Closeable { outputStream.onError(new RuntimeException(responseMsg.getErrorMsg())); } else { if (requestMsg.getConnectRequestMsg().hasMaxInboundMessageSize()) { - log.debug("[{}] Client max inbound message size: {}", sessionId, requestMsg.getConnectRequestMsg().getMaxInboundMessageSize()); + log.debug("[{}][{}] Client max inbound message size: {}", tenantId, sessionId, requestMsg.getConnectRequestMsg().getMaxInboundMessageSize()); clientMaxInboundMessageSize = requestMsg.getConnectRequestMsg().getMaxInboundMessageSize(); } connected = true; @@ -179,13 +180,13 @@ public final class EdgeGrpcSession implements Closeable { @Override public void onError(Throwable t) { - log.error("[{}] Stream was terminated due to error:", sessionId, t); + log.error("[{}][{}] Stream was terminated due to error:", tenantId, sessionId, t); closeSession(); } @Override public void onCompleted() { - log.info("[{}] Stream was closed and completed successfully!", sessionId); + log.info("[{}][{}] Stream was closed and completed successfully!", tenantId, sessionId); closeSession(); } @@ -206,7 +207,7 @@ public final class EdgeGrpcSession implements Closeable { } public void startSyncProcess(boolean fullSync) { - log.trace("[{}][{}][{}] Staring edge sync process", edge.getTenantId(), edge.getId(), this.sessionId); + log.trace("[{}][{}][{}] Staring edge sync process", this.tenantId, edge.getId(), this.sessionId); syncCompleted = false; interruptGeneralProcessingOnSync(); doSync(new EdgeSyncCursor(ctx, edge, fullSync)); @@ -216,7 +217,7 @@ public final class EdgeGrpcSession implements Closeable { if (cursor.hasNext()) { EdgeEventFetcher next = cursor.getNext(); log.info("[{}][{}] starting sync process, cursor current idx = {}, class = {}", - edge.getTenantId(), edge.getId(), cursor.getCurrentIdx(), next.getClass().getSimpleName()); + this.tenantId, edge.getId(), cursor.getCurrentIdx(), next.getClass().getSimpleName()); ListenableFuture> future = startProcessingEdgeEvents(next); Futures.addCallback(future, new FutureCallback<>() { @Override @@ -226,7 +227,7 @@ public final class EdgeGrpcSession implements Closeable { @Override public void onFailure(Throwable t) { - log.error("[{}][{}] Exception during sync process", edge.getTenantId(), edge.getId(), t); + log.error("[{}][{}] Exception during sync process", tenantId, edge.getId(), t); } }, ctx.getGrpcCallbackExecutorService()); } else { @@ -243,7 +244,7 @@ public final class EdgeGrpcSession implements Closeable { @Override public void onFailure(Throwable t) { - log.error("[{}][{}] Exception during sending sync complete", edge.getTenantId(), edge.getId(), t); + log.error("[{}][{}] Exception during sending sync complete", tenantId, edge.getId(), t); } }, ctx.getGrpcCallbackExecutorService()); } @@ -279,39 +280,40 @@ public final class EdgeGrpcSession implements Closeable { try { if (msg.getSuccess()) { sessionState.getPendingMsgsMap().remove(msg.getDownlinkMsgId()); - log.debug("[{}] Msg has been processed successfully!Msd Id: [{}], Msg: {}", edge.getRoutingKey(), msg.getDownlinkMsgId(), msg); + log.debug("[{}][{}] Msg has been processed successfully!Msd Id: [{}], Msg: {}", this.tenantId, edge.getRoutingKey(), msg.getDownlinkMsgId(), msg); } else { - log.error("[{}] Msg processing failed! Msd Id: [{}], Error msg: {}", edge.getRoutingKey(), msg.getDownlinkMsgId(), msg.getErrorMsg()); + log.error("[{}][{}] Msg processing failed! Msd Id: [{}], Error msg: {}", this.tenantId, edge.getRoutingKey(), msg.getDownlinkMsgId(), msg.getErrorMsg()); } if (sessionState.getPendingMsgsMap().isEmpty()) { - log.debug("[{}] Pending msgs map is empty. Stopping current iteration", edge.getRoutingKey()); + log.debug("[{}][{}] Pending msgs map is empty. Stopping current iteration", this.tenantId, edge.getRoutingKey()); stopCurrentSendDownlinkMsgsTask(false); } } catch (Exception e) { - log.error("[{}] Can't process downlink response message [{}]", this.sessionId, msg, e); + log.error("[{}][{}] Can't process downlink response message [{}]", this.tenantId, this.sessionId, msg, e); } } private void sendDownlinkMsg(ResponseMsg downlinkMsg) { - log.trace("[{}] Sending downlink msg [{}]", this.sessionId, downlinkMsg); + log.trace("[{}][{}] Sending downlink msg [{}]", this.tenantId, this.sessionId, downlinkMsg); if (isConnected()) { downlinkMsgLock.lock(); try { outputStream.onNext(downlinkMsg); } catch (Exception e) { - log.error("[{}] Failed to send downlink message [{}]", this.sessionId, downlinkMsg, e); + log.error("[{}][{}] Failed to send downlink message [{}]", this.tenantId, this.sessionId, downlinkMsg, e); connected = false; sessionCloseListener.accept(edge.getId(), sessionId); } finally { downlinkMsgLock.unlock(); } - log.trace("[{}] Response msg successfully sent [{}]", this.sessionId, downlinkMsg); + log.trace("[{}][{}] Response msg successfully sent [{}]", this.tenantId, this.sessionId, downlinkMsg); } } void onConfigurationUpdate(Edge edge) { log.debug("[{}] onConfigurationUpdate [{}]", this.sessionId, edge); this.edge = edge; + this.tenantId = edge.getTenantId(); EdgeUpdateMsg edgeConfig = EdgeUpdateMsg.newBuilder() .setConfiguration(ctx.getEdgeMsgConstructor().constructEdgeConfiguration(edge)).build(); ResponseMsg edgeConfigMsg = ResponseMsg.newBuilder() @@ -322,7 +324,7 @@ public final class EdgeGrpcSession implements Closeable { ListenableFuture processEdgeEvents() throws Exception { SettableFuture result = SettableFuture.create(); - log.trace("[{}] starting processing edge events", this.sessionId); + log.trace("[{}][{}] starting processing edge events", this.tenantId, this.sessionId); if (isConnected() && isSyncCompleted()) { Pair startTsAndSeqId = getQueueStartTsAndSeqId().get(); this.previousStartTs = startTsAndSeqId.getFirst(); @@ -342,7 +344,7 @@ public final class EdgeGrpcSession implements Closeable { Futures.addCallback(updateFuture, new FutureCallback<>() { @Override public void onSuccess(@Nullable List list) { - log.debug("[{}] queue offset was updated [{}]", sessionId, newStartTsAndSeqId); + log.debug("[{}][{}] queue offset was updated [{}]", tenantId, sessionId, newStartTsAndSeqId); if (fetcher.isSeqIdNewCycleStarted()) { seqIdEnd = fetcher.getSeqIdEnd(); boolean newEventsAvailable = isNewEdgeEventsAvailable(); @@ -359,24 +361,24 @@ public final class EdgeGrpcSession implements Closeable { @Override public void onFailure(Throwable t) { - log.error("[{}] Failed to update queue offset [{}]", sessionId, newStartTsAndSeqId, t); + log.error("[{}][{}] Failed to update queue offset [{}]", tenantId, sessionId, newStartTsAndSeqId, t); result.setException(t); } }, ctx.getGrpcCallbackExecutorService()); } else { - log.trace("[{}] newStartTsAndSeqId is null. Skipping iteration without db update", sessionId); + log.trace("[{}][{}] newStartTsAndSeqId is null. Skipping iteration without db update", tenantId, sessionId); result.set(null); } } @Override public void onFailure(Throwable t) { - log.error("[{}] Failed to process events", sessionId, t); + log.error("[{}][{}] Failed to process events", tenantId, sessionId, t); result.setException(t); } }, ctx.getGrpcCallbackExecutorService()); } else { - log.trace("[{}] edge is not connected or sync is not completed. Skipping iteration", sessionId); + log.trace("[{}][{}] edge is not connected or sync is not completed. Skipping iteration", tenantId, sessionId); result.set(null); } return result; @@ -393,13 +395,13 @@ public final class EdgeGrpcSession implements Closeable { try { PageData pageData = fetcher.fetchEdgeEvents(edge.getTenantId(), edge, pageLink); if (isConnected() && !pageData.getData().isEmpty()) { - log.trace("[{}] [{}] event(s) are going to be processed.", this.sessionId, pageData.getData().size()); + log.trace("[{}][{}][{}] event(s) are going to be processed.", this.tenantId, this.sessionId, pageData.getData().size()); List downlinkMsgsPack = convertToDownlinkMsgsPack(pageData.getData()); Futures.addCallback(sendDownlinkMsgsPack(downlinkMsgsPack), new FutureCallback<>() { @Override public void onSuccess(@Nullable Boolean isInterrupted) { if (Boolean.TRUE.equals(isInterrupted)) { - log.debug("[{}][{}][{}] Send downlink messages task was interrupted", edge.getTenantId(), edge.getId(), sessionId); + log.debug("[{}][{}][{}] Send downlink messages task was interrupted", tenantId, edge.getId(), sessionId); result.set(null); } else { if (isConnected() && pageData.hasNext()) { @@ -452,14 +454,14 @@ public final class EdgeGrpcSession implements Closeable { if (isConnected() && sessionState.getPendingMsgsMap().values().size() > 0) { List copy = new ArrayList<>(sessionState.getPendingMsgsMap().values()); if (attempt > 1) { - log.warn("[{}] Failed to deliver the batch: {}, attempt: {}", this.sessionId, copy, attempt); + log.warn("[{}][{}] Failed to deliver the batch: {}, attempt: {}", this.tenantId, this.sessionId, copy, attempt); } - log.trace("[{}] [{}] downlink msg(s) are going to be send.", this.sessionId, copy.size()); + log.trace("[{}][{}][{}] downlink msg(s) are going to be send.", this.tenantId, this.sessionId, copy.size()); for (DownlinkMsg downlinkMsg : copy) { if (this.clientMaxInboundMessageSize != 0 && downlinkMsg.getSerializedSize() > this.clientMaxInboundMessageSize) { log.error("[{}][{}][{}] Downlink msg size [{}] exceeds client max inbound message size [{}]. Skipping this message. " + "Please increase value of CLOUD_RPC_MAX_INBOUND_MESSAGE_SIZE env variable on the edge and restart it." + - "Message {}", edge.getTenantId(), edge.getId(), this.sessionId, downlinkMsg.getSerializedSize(), + "Message {}", this.tenantId, edge.getId(), this.sessionId, downlinkMsg.getSerializedSize(), this.clientMaxInboundMessageSize, downlinkMsg); sessionState.getPendingMsgsMap().remove(downlinkMsg.getDownlinkMsgId()); } else { @@ -471,15 +473,15 @@ public final class EdgeGrpcSession implements Closeable { if (attempt < MAX_DOWNLINK_ATTEMPTS) { scheduleDownlinkMsgsPackSend(attempt + 1); } else { - log.warn("[{}] Failed to deliver the batch after {} attempts. Next messages are going to be discarded {}", - this.sessionId, MAX_DOWNLINK_ATTEMPTS, copy); + log.warn("[{}][{}] Failed to deliver the batch after {} attempts. Next messages are going to be discarded {}", + this.tenantId, this.sessionId, MAX_DOWNLINK_ATTEMPTS, copy); stopCurrentSendDownlinkMsgsTask(false); } } else { stopCurrentSendDownlinkMsgsTask(false); } } catch (Exception e) { - log.warn("[{}] Failed to send downlink msgs. Error msg {}", this.sessionId, e.getMessage(), e); + log.warn("[{}][{}] Failed to send downlink msgs. Error msg {}", this.tenantId, this.sessionId, e.getMessage(), e); stopCurrentSendDownlinkMsgsTask(true); } }; @@ -499,7 +501,7 @@ public final class EdgeGrpcSession implements Closeable { private List convertToDownlinkMsgsPack(List edgeEvents) { List result = new ArrayList<>(); for (EdgeEvent edgeEvent : edgeEvents) { - log.trace("[{}][{}] converting edge event to downlink msg [{}]", edge.getTenantId(), this.sessionId, edgeEvent); + log.trace("[{}][{}] converting edge event to downlink msg [{}]", this.tenantId, this.sessionId, edgeEvent); DownlinkMsg downlinkMsg = null; try { switch (edgeEvent.getAction()) { @@ -518,7 +520,7 @@ public final class EdgeGrpcSession implements Closeable { case ASSIGNED_TO_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER: downlinkMsg = convertEntityEventToDownlink(edgeEvent); - log.trace("[{}][{}] entity message processed [{}]", edgeEvent.getTenantId(), this.sessionId, downlinkMsg); + log.trace("[{}][{}] entity message processed [{}]", this.tenantId, this.sessionId, downlinkMsg); break; case ATTRIBUTES_UPDATED: case POST_ATTRIBUTES: @@ -527,10 +529,10 @@ public final class EdgeGrpcSession implements Closeable { downlinkMsg = ctx.getTelemetryProcessor().convertTelemetryEventToDownlink(edgeEvent); break; default: - log.warn("[{}][{}] Unsupported action type [{}]", edge.getTenantId(), this.sessionId, edgeEvent.getAction()); + log.warn("[{}][{}] Unsupported action type [{}]", this.tenantId, this.sessionId, edgeEvent.getAction()); } } catch (Exception e) { - log.error("[{}][{}] Exception during converting edge event to downlink msg", edge.getTenantId(), this.sessionId, e); + log.error("[{}][{}] Exception during converting edge event to downlink msg", this.tenantId, this.sessionId, e); } if (downlinkMsg != null) { result.add(downlinkMsg); @@ -566,7 +568,7 @@ public final class EdgeGrpcSession implements Closeable { PageData edgeEvents = ctx.getEdgeEventService().findEdgeEvents(edge.getTenantId(), edge.getId(), 0L, this.previousStartSeqId == 0 ? null : this.previousStartSeqId - 1, pageLink); return !edgeEvents.getData().isEmpty(); } catch (Exception e) { - log.error("[{}][{}][{}] Failed to execute isSeqIdStartedNewCycle", edge.getTenantId(), edge.getId(), sessionId, e); + log.error("[{}][{}][{}] Failed to execute isSeqIdStartedNewCycle", this.tenantId, edge.getId(), sessionId, e); } return false; } @@ -577,7 +579,7 @@ public final class EdgeGrpcSession implements Closeable { PageData edgeEvents = ctx.getEdgeEventService().findEdgeEvents(edge.getTenantId(), edge.getId(), this.newStartSeqId, null, pageLink); return !edgeEvents.getData().isEmpty(); } catch (Exception e) { - log.error("[{}][{}][{}] Failed to execute isNewEdgeEventsAvailable", edge.getTenantId(), edge.getId(), sessionId, e); + log.error("[{}][{}][{}] Failed to execute isNewEdgeEventsAvailable", this.tenantId, edge.getId(), sessionId, e); } return false; } @@ -591,7 +593,7 @@ public final class EdgeGrpcSession implements Closeable { startSeqId = edgeEvents.getData().get(0).getSeqId() - 1; } } catch (Exception e) { - log.error("[{}][{}][{}] Failed to execute findStartSeqIdFromOldestEventIfAny", edge.getTenantId(), edge.getId(), sessionId, e); + log.error("[{}][{}][{}] Failed to execute findStartSeqIdFromOldestEventIfAny", this.tenantId, edge.getId(), sessionId, e); } return startSeqId; } @@ -607,7 +609,7 @@ public final class EdgeGrpcSession implements Closeable { } private DownlinkMsg convertEntityEventToDownlink(EdgeEvent edgeEvent) { - log.trace("Executing convertEntityEventToDownlink, edgeEvent [{}], action [{}]", edgeEvent, edgeEvent.getAction()); + log.trace("[{}] Executing convertEntityEventToDownlink, edgeEvent [{}], action [{}]", this.tenantId, edgeEvent, edgeEvent.getAction()); switch (edgeEvent.getType()) { case EDGE: return ctx.getEdgeProcessor().convertEdgeEventToDownlink(edgeEvent); @@ -650,7 +652,7 @@ public final class EdgeGrpcSession implements Closeable { case TENANT_PROFILE: return ctx.getTenantProfileEdgeProcessor().convertTenantProfileEventToDownlink(edgeEvent); default: - log.warn("Unsupported edge event type [{}]", edgeEvent); + log.warn("[{}] Unsupported edge event type [{}]", this.tenantId, edgeEvent); return null; } } @@ -749,7 +751,7 @@ public final class EdgeGrpcSession implements Closeable { } } } catch (Exception e) { - log.error("[{}] Can't process uplink msg [{}]", this.sessionId, uplinkMsg, e); + log.error("[{}][{}] Can't process uplink msg [{}]", this.tenantId, this.sessionId, uplinkMsg, e); return Futures.immediateFailedFuture(e); } return Futures.allAsList(result); @@ -791,25 +793,25 @@ public final class EdgeGrpcSession implements Closeable { @Override public void close() { - log.debug("[{}] Closing session", sessionId); + log.debug("[{}][{}] Closing session", this.tenantId, sessionId); connected = false; try { outputStream.onCompleted(); } catch (Exception e) { - log.debug("[{}] Failed to close output stream: {}", sessionId, e.getMessage()); + log.debug("[{}][{}] Failed to close output stream: {}", this.tenantId, sessionId, e.getMessage()); } } private void interruptPreviousSendDownlinkMsgsTask() { if (sessionState.getSendDownlinkMsgsFuture() != null && !sessionState.getSendDownlinkMsgsFuture().isDone() || sessionState.getScheduledSendDownlinkTask() != null && !sessionState.getScheduledSendDownlinkTask().isCancelled()) { - log.debug("[{}][{}][{}] Previous send downlink future was not properly completed, stopping it now!", edge.getTenantId(), edge.getId(), this.sessionId); + log.debug("[{}][{}][{}] Previous send downlink future was not properly completed, stopping it now!", this.tenantId, edge.getId(), this.sessionId); stopCurrentSendDownlinkMsgsTask(true); } } private void interruptGeneralProcessingOnSync() { - log.debug("[{}][{}][{}] Sync process started. General processing interrupted!", edge.getTenantId(), edge.getId(), this.sessionId); + log.debug("[{}][{}][{}] Sync process started. General processing interrupted!", this.tenantId, edge.getId(), this.sessionId); stopCurrentSendDownlinkMsgsTask(true); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AlarmMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AlarmMsgConstructor.java index 447a73e5cf..15d7c27ff6 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AlarmMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AlarmMsgConstructor.java @@ -15,20 +15,9 @@ */ package org.thingsboard.server.service.edge.rpc.constructor; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.thingsboard.common.util.JacksonUtil; -import org.thingsboard.server.common.data.Device; -import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.alarm.Alarm; -import org.thingsboard.server.common.data.asset.Asset; -import org.thingsboard.server.common.data.id.AssetId; -import org.thingsboard.server.common.data.id.DeviceId; -import org.thingsboard.server.common.data.id.EntityViewId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.dao.asset.AssetService; -import org.thingsboard.server.dao.device.DeviceService; -import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.gen.edge.v1.AlarmUpdateMsg; import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -37,37 +26,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent; @TbCoreComponent public class AlarmMsgConstructor { - @Autowired - private DeviceService deviceService; - - @Autowired - private AssetService assetService; - - @Autowired - private EntityViewService entityViewService; - - public AlarmUpdateMsg constructAlarmUpdatedMsg(TenantId tenantId, UpdateMsgType msgType, Alarm alarm) { - String entityName = null; - switch (alarm.getOriginator().getEntityType()) { - case DEVICE: - Device deviceById = deviceService.findDeviceById(tenantId, new DeviceId(alarm.getOriginator().getId())); - if (deviceById != null) { - entityName = deviceById.getName(); - } - break; - case ASSET: - Asset assetById = assetService.findAssetById(tenantId, new AssetId(alarm.getOriginator().getId())); - if (assetById != null) { - entityName = assetById.getName(); - } - break; - case ENTITY_VIEW: - EntityView entityViewById = entityViewService.findEntityViewById(tenantId, new EntityViewId(alarm.getOriginator().getId())); - if (entityViewById != null) { - entityName = entityViewById.getName(); - } - break; - } + public AlarmUpdateMsg constructAlarmUpdatedMsg(UpdateMsgType msgType, Alarm alarm, String entityName) { AlarmUpdateMsg.Builder builder = AlarmUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(alarm.getId().getId().getMostSignificantBits()) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java index 242a15ba56..47dbad231d 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java @@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.transport.adaptor.JsonConverter; import org.thingsboard.server.gen.edge.v1.AttributeDeleteMsg; import org.thingsboard.server.gen.edge.v1.EntityDataProto; @@ -40,7 +41,7 @@ import java.util.List; @TbCoreComponent public class EntityDataMsgConstructor { - public EntityDataProto constructEntityDataMsg(EntityId entityId, EdgeEventActionType actionType, JsonElement entityData) { + public EntityDataProto constructEntityDataMsg(TenantId tenantId, EntityId entityId, EdgeEventActionType actionType, JsonElement entityData) { EntityDataProto.Builder builder = EntityDataProto.newBuilder() .setEntityIdMSB(entityId.getId().getMostSignificantBits()) .setEntityIdLSB(entityId.getId().getLeastSignificantBits()) @@ -57,7 +58,7 @@ public class EntityDataMsgConstructor { } builder.setPostTelemetryMsg(JsonConverter.convertToTelemetryProto(data.getAsJsonObject("data"), ts)); } catch (Exception e) { - log.warn("[{}] Can't convert to telemetry proto, entityData [{}]", entityId, entityData, e); + log.warn("[{}][{}] Can't convert to telemetry proto, entityData [{}]", tenantId, entityId, entityData, e); } break; case ATTRIBUTES_UPDATED: @@ -67,7 +68,7 @@ public class EntityDataMsgConstructor { builder.setAttributesUpdatedMsg(attributesUpdatedMsg); builder.setPostAttributeScope(getScopeOfDefault(data)); } catch (Exception e) { - log.warn("[{}] Can't convert to AttributesUpdatedMsg proto, entityData [{}]", entityId, entityData, e); + log.warn("[{}][{}] Can't convert to AttributesUpdatedMsg proto, entityData [{}]", tenantId, entityId, entityData, e); } break; case POST_ATTRIBUTES: @@ -77,7 +78,7 @@ public class EntityDataMsgConstructor { builder.setPostAttributesMsg(postAttributesMsg); builder.setPostAttributeScope(getScopeOfDefault(data)); } catch (Exception e) { - log.warn("[{}] Can't convert to PostAttributesMsg, entityData [{}]", entityId, entityData, e); + log.warn("[{}][{}] Can't convert to PostAttributesMsg, entityData [{}]", tenantId, entityId, entityData, e); } break; case ATTRIBUTES_DELETED: @@ -90,7 +91,7 @@ public class EntityDataMsgConstructor { attributeDeleteMsg.build(); builder.setAttributeDeleteMsg(attributeDeleteMsg); } catch (Exception e) { - log.warn("[{}] Can't convert to AttributeDeleteMsg proto, entityData [{}]", entityId, entityData, e); + log.warn("[{}][{}] Can't convert to AttributeDeleteMsg proto, entityData [{}]", tenantId, entityId, entityData, e); } break; } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/rule/AbstractRuleChainMetadataConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/rule/AbstractRuleChainMetadataConstructor.java index 28c1b1c458..0099d85bcd 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/rule/AbstractRuleChainMetadataConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/rule/AbstractRuleChainMetadataConstructor.java @@ -51,7 +51,7 @@ public abstract class AbstractRuleChainMetadataConstructor implements RuleChainM builder.setMsgType(msgType); return builder.build(); } catch (JsonProcessingException ex) { - log.error("Can't construct RuleChainMetadataUpdateMsg", ex); + log.error("[{}] Can't construct RuleChainMetadataUpdateMsg", tenantId, ex); } return null; } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/AdminSettingsEdgeEventFetcher.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/AdminSettingsEdgeEventFetcher.java index cefeb4c46a..b9fc1c8c9e 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/AdminSettingsEdgeEventFetcher.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/AdminSettingsEdgeEventFetcher.java @@ -85,7 +85,7 @@ public class AdminSettingsEdgeEventFetcher implements EdgeEventFetcher { result.add(EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, JacksonUtil.OBJECT_MAPPER.valueToTree(tenantMailSettings))); - AdminSettings systemMailTemplates = loadMailTemplates(); + AdminSettings systemMailTemplates = loadMailTemplates(tenantId); result.add(EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, JacksonUtil.OBJECT_MAPPER.valueToTree(systemMailTemplates))); @@ -97,7 +97,7 @@ public class AdminSettingsEdgeEventFetcher implements EdgeEventFetcher { return new PageData<>(result, 1, result.size(), false); } - private AdminSettings loadMailTemplates() throws Exception { + private AdminSettings loadMailTemplates(TenantId tenantId) throws Exception { Map mailTemplates = new HashMap<>(); for (String templatesName : templatesNames) { Template template = freemarkerConfig.getTemplate(templatesName); @@ -107,7 +107,7 @@ public class AdminSettingsEdgeEventFetcher implements EdgeEventFetcher { if (mailTemplate != null) { mailTemplates.put(name, mailTemplate); } else { - log.error("Can't load mail template from file {}", template.getName()); + log.error("[{}] Can't load mail template from file {}", tenantId, template.getName()); } } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/BaseAlarmProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/BaseAlarmProcessor.java index fbe141ab32..de072a48e9 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/BaseAlarmProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/BaseAlarmProcessor.java @@ -20,15 +20,21 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.alarm.AlarmCreateOrUpdateActiveRequest; import org.thingsboard.server.common.data.alarm.AlarmSeverity; import org.thingsboard.server.common.data.alarm.AlarmStatus; import org.thingsboard.server.common.data.alarm.AlarmUpdateRequest; +import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.id.AlarmId; +import org.thingsboard.server.common.data.id.AssetId; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.gen.edge.v1.AlarmUpdateMsg; import org.thingsboard.server.gen.edge.v1.UpdateMsgType; @@ -45,7 +51,7 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor { EntityType.valueOf(alarmUpdateMsg.getOriginatorType())); AlarmId alarmId = new AlarmId(new UUID(alarmUpdateMsg.getIdMSB(), alarmUpdateMsg.getIdLSB())); if (originatorId == null) { - log.warn("Originator not found for the alarm msg {}", alarmUpdateMsg); + log.warn("[{}] Originator not found for the alarm msg {}", tenantId, alarmUpdateMsg); return Futures.immediateFuture(null); } try { @@ -129,13 +135,38 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor { case ALARM_CLEAR: Alarm alarm = alarmService.findAlarmById(tenantId, alarmId); if (alarm != null) { - return alarmMsgConstructor.constructAlarmUpdatedMsg(tenantId, msgType, alarm); + return alarmMsgConstructor.constructAlarmUpdatedMsg(msgType, alarm, findOriginatorEntityName(tenantId, alarm)); } break; case DELETED: Alarm deletedAlarm = JacksonUtil.OBJECT_MAPPER.convertValue(body, Alarm.class); - return alarmMsgConstructor.constructAlarmUpdatedMsg(tenantId, msgType, deletedAlarm); + return alarmMsgConstructor.constructAlarmUpdatedMsg(msgType, deletedAlarm, findOriginatorEntityName(tenantId, deletedAlarm)); } return null; } + + private String findOriginatorEntityName(TenantId tenantId, Alarm alarm) { + String entityName = null; + switch (alarm.getOriginator().getEntityType()) { + case DEVICE: + Device deviceById = deviceService.findDeviceById(tenantId, new DeviceId(alarm.getOriginator().getId())); + if (deviceById != null) { + entityName = deviceById.getName(); + } + break; + case ASSET: + Asset assetById = assetService.findAssetById(tenantId, new AssetId(alarm.getOriginator().getId())); + if (assetById != null) { + entityName = assetById.getName(); + } + break; + case ENTITY_VIEW: + EntityView entityViewById = entityViewService.findEntityViewById(tenantId, new EntityViewId(alarm.getOriginator().getId())); + if (entityViewById != null) { + entityName = entityViewById.getName(); + } + break; + } + return entityName; + } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/AssetProfileEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/AssetProfileEdgeProcessor.java index a40cd64ccd..21865baa09 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/AssetProfileEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/AssetProfileEdgeProcessor.java @@ -69,7 +69,7 @@ public class AssetProfileEdgeProcessor extends BaseAssetProfileProcessor { return handleUnsupportedMsgType(assetProfileUpdateMsg.getMsgType()); } } catch (DataValidationException e) { - log.warn("Failed to process AssetProfileUpdateMsg from Edge [{}]", assetProfileUpdateMsg, e); + log.warn("[{}] Failed to process AssetProfileUpdateMsg from Edge [{}]", tenantId, assetProfileUpdateMsg, e); return Futures.immediateFailedFuture(e); } finally { edgeSynchronizationManager.getSync().remove(); @@ -98,16 +98,16 @@ public class AssetProfileEdgeProcessor extends BaseAssetProfileProcessor { tbClusterService.pushMsgToRuleEngine(tenantId, assetProfileId, tbMsg, new TbQueueCallback() { @Override public void onSuccess(TbQueueMsgMetadata metadata) { - log.debug("Successfully send ENTITY_CREATED EVENT to rule engine [{}]", assetProfile); + log.debug("[{}] Successfully send ENTITY_CREATED EVENT to rule engine [{}]", tenantId, assetProfile); } @Override public void onFailure(Throwable t) { - log.warn("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", assetProfile, t); + log.warn("[{}] Failed to send ENTITY_CREATED EVENT to rule engine [{}]", tenantId, assetProfile, t); } }); } catch (JsonProcessingException | IllegalArgumentException e) { - log.warn("[{}] Failed to push asset profile action to rule engine: {}", assetProfileId, DataConstants.ENTITY_CREATED, e); + log.warn("[{}][{}] Failed to push asset profile action to rule engine: {}", tenantId, assetProfileId, DataConstants.ENTITY_CREATED, e); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProcessor.java index 5d728cb4e8..f8dcdbd8f6 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProcessor.java @@ -49,8 +49,8 @@ public abstract class BaseAssetProcessor extends BaseEdgeProcessor { Asset assetByName = assetService.findAssetByTenantIdAndName(tenantId, assetName); if (assetByName != null && !assetByName.getId().equals(assetId)) { assetName = assetName + "_" + StringUtils.randomAlphanumeric(15); - log.warn("Asset with name {} already exists. Renaming asset name to {}", - assetUpdateMsg.getName(), assetName); + log.warn("[{}] Asset with name {} already exists. Renaming asset name to {}", + tenantId, assetUpdateMsg.getName(), assetName); assetNameUpdated = true; } asset.setName(assetName); @@ -69,6 +69,9 @@ public abstract class BaseAssetProcessor extends BaseEdgeProcessor { asset.setId(assetId); } assetService.saveAsset(asset, false); + } catch (Exception e) { + log.error("[{}] Failed to process asset update msg [{}]", tenantId, assetUpdateMsg, e); + throw e; } finally { assetCreationLock.unlock(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProfileProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProfileProcessor.java index 5e35bf0332..67c8f8de44 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProfileProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProfileProcessor.java @@ -46,8 +46,8 @@ public abstract class BaseAssetProfileProcessor extends BaseEdgeProcessor { AssetProfile assetProfileByName = assetProfileService.findAssetProfileByName(tenantId, assetProfileName); if (assetProfileByName != null && !assetProfileByName.getId().equals(assetProfileId)) { assetProfileName = assetProfileName + "_" + StringUtils.randomAlphabetic(15); - log.warn("Asset profile with name {} already exists. Renaming asset profile name to {}", - assetProfileUpdateMsg.getName(), assetProfileName); + log.warn("[{}] Asset profile with name {} already exists. Renaming asset profile name to {}", + tenantId, assetProfileUpdateMsg.getName(), assetProfileName); assetProfileNameUpdated = true; } assetProfile.setName(assetProfileName); @@ -66,6 +66,9 @@ public abstract class BaseAssetProfileProcessor extends BaseEdgeProcessor { assetProfile.setId(assetProfileId); } assetProfileService.saveAssetProfile(assetProfile, false); + } catch (Exception e) { + log.error("[{}] Failed to process asset profile update msg [{}]", tenantId, assetProfileUpdateMsg, e); + throw e; } finally { assetCreationLock.unlock(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProcessor.java index 580f752ac1..7993eb4fd8 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProcessor.java @@ -61,8 +61,8 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor { Device deviceByName = deviceService.findDeviceByTenantIdAndName(tenantId, deviceName); if (deviceByName != null && !deviceByName.getId().equals(deviceId)) { deviceName = deviceName + "_" + StringUtils.randomAlphabetic(15); - log.warn("Device with name {} already exists. Renaming device name to {}", - deviceUpdateMsg.getName(), deviceName); + log.warn("[{}] Device with name {} already exists. Renaming device name to {}", + tenantId, deviceUpdateMsg.getName(), deviceName); deviceNameUpdated = true; } device.setName(deviceName); @@ -98,7 +98,10 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor { deviceCredentialsService.createDeviceCredentials(device.getTenantId(), deviceCredentials); } tbClusterService.onDeviceUpdated(savedDevice, created ? null : device); - } finally { + } catch (Exception e) { + log.error("[{}] Failed to process device update msg [{}]", tenantId, deviceUpdateMsg, e); + throw e; + } finally { deviceCreationLock.unlock(); } return Pair.of(created, deviceNameUpdated); @@ -110,8 +113,8 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor { return dbCallbackExecutorService.submit(() -> { Device device = deviceService.findDeviceById(tenantId, deviceId); if (device != null) { - log.debug("Updating device credentials for device [{}]. New device credentials Id [{}], value [{}]", - device.getName(), deviceCredentialsUpdateMsg.getCredentialsId(), deviceCredentialsUpdateMsg.getCredentialsValue()); + log.debug("[{}] Updating device credentials for device [{}]. New device credentials Id [{}], value [{}]", + tenantId, device.getName(), deviceCredentialsUpdateMsg.getCredentialsId(), deviceCredentialsUpdateMsg.getCredentialsValue()); try { edgeSynchronizationManager.getSync().set(true); @@ -122,14 +125,14 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor { ? deviceCredentialsUpdateMsg.getCredentialsValue() : null); deviceCredentialsService.updateDeviceCredentials(tenantId, deviceCredentials); } catch (Exception e) { - log.error("Can't update device credentials for device [{}], deviceCredentialsUpdateMsg [{}]", - device.getName(), deviceCredentialsUpdateMsg, e); + log.error("[{}] Can't update device credentials for device [{}], deviceCredentialsUpdateMsg [{}]", + tenantId, device.getName(), deviceCredentialsUpdateMsg, e); throw new RuntimeException(e); } finally { edgeSynchronizationManager.getSync().remove(); } } else { - log.warn("Can't find device by id [{}], deviceCredentialsUpdateMsg [{}]", deviceId, deviceCredentialsUpdateMsg); + log.warn("[{}] Can't find device by id [{}], deviceCredentialsUpdateMsg [{}]", tenantId, deviceId, deviceCredentialsUpdateMsg); } return null; }); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProfileProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProfileProcessor.java index 324cf14584..bd35a42e59 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProfileProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProfileProcessor.java @@ -58,8 +58,8 @@ public abstract class BaseDeviceProfileProcessor extends BaseEdgeProcessor { DeviceProfile deviceProfileByName = deviceProfileService.findDeviceProfileByName(tenantId, deviceProfileName); if (deviceProfileByName != null && !deviceProfileByName.getId().equals(deviceProfileId)) { deviceProfileName = deviceProfileName + "_" + StringUtils.randomAlphabetic(15); - log.warn("Device profile with name {} already exists. Renaming device profile name to {}", - deviceProfileUpdateMsg.getName(), deviceProfileName); + log.warn("[{}] Device profile with name {} already exists. Renaming device profile name to {}", + tenantId, deviceProfileUpdateMsg.getName(), deviceProfileName); deviceProfileNameUpdated = true; } deviceProfile.setName(deviceProfileName); @@ -98,7 +98,10 @@ public abstract class BaseDeviceProfileProcessor extends BaseEdgeProcessor { deviceProfile.setId(deviceProfileId); } deviceProfileService.saveDeviceProfile(deviceProfile, false); - } finally { + } catch (Exception e) { + log.error("[{}] Failed to process device profile update msg [{}]", tenantId, deviceProfileUpdateMsg, e); + throw e; + } finally { deviceCreationLock.unlock(); } return Pair.of(created, deviceProfileNameUpdated); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceProfileEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceProfileEdgeProcessor.java index 470e396cb2..fe126afd6d 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceProfileEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceProfileEdgeProcessor.java @@ -52,7 +52,6 @@ import java.util.UUID; @TbCoreComponent public class DeviceProfileEdgeProcessor extends BaseDeviceProfileProcessor { - public ListenableFuture processDeviceProfileMsgFromEdge(TenantId tenantId, Edge edge, DeviceProfileUpdateMsg deviceProfileUpdateMsg) { log.trace("[{}] executing processDeviceProfileMsgFromEdge [{}] from edge [{}]", tenantId, deviceProfileUpdateMsg, edge.getName()); DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(deviceProfileUpdateMsg.getIdMSB(), deviceProfileUpdateMsg.getIdLSB())); @@ -70,7 +69,7 @@ public class DeviceProfileEdgeProcessor extends BaseDeviceProfileProcessor { return handleUnsupportedMsgType(deviceProfileUpdateMsg.getMsgType()); } } catch (DataValidationException e) { - log.warn("Failed to process DeviceProfileUpdateMsg from Edge [{}]", deviceProfileUpdateMsg, e); + log.warn("[{}] Failed to process DeviceProfileUpdateMsg from Edge [{}]", tenantId, deviceProfileUpdateMsg, e); return Futures.immediateFailedFuture(e); } finally { edgeSynchronizationManager.getSync().remove(); @@ -99,16 +98,16 @@ public class DeviceProfileEdgeProcessor extends BaseDeviceProfileProcessor { tbClusterService.pushMsgToRuleEngine(tenantId, deviceProfileId, tbMsg, new TbQueueCallback() { @Override public void onSuccess(TbQueueMsgMetadata metadata) { - log.debug("Successfully send ENTITY_CREATED EVENT to rule engine [{}]", deviceProfile); + log.debug("[{}] Successfully send ENTITY_CREATED EVENT to rule engine [{}]", tenantId, deviceProfile); } @Override public void onFailure(Throwable t) { - log.warn("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", deviceProfile, t); + log.warn("[{}] Failed to send ENTITY_CREATED EVENT to rule engine [{}]", tenantId, deviceProfile, t); } }); } catch (JsonProcessingException | IllegalArgumentException e) { - log.warn("[{}] Failed to push device profile action to rule engine: {}", deviceProfileId, DataConstants.ENTITY_CREATED, e); + log.warn("[{}][{}] Failed to push device profile action to rule engine: {}", tenantId, deviceProfileId, DataConstants.ENTITY_CREATED, e); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeProcessor.java index 123209afc1..b38659c881 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeProcessor.java @@ -85,7 +85,7 @@ public class EdgeProcessor extends BaseEdgeProcessor { do { pageData = userService.findCustomerUsers(tenantId, customerId, pageLink); if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { - log.trace("[{}] [{}] user(s) are going to be added to edge.", edge.getId(), pageData.getData().size()); + log.trace("[{}][{}][{}] user(s) are going to be added to edge.", tenantId, edge.getId(), pageData.getData().size()); for (User user : pageData.getData()) { futures.add(saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, EdgeEventActionType.ADDED, user.getId(), null)); } @@ -108,7 +108,7 @@ public class EdgeProcessor extends BaseEdgeProcessor { return Futures.immediateFuture(null); } } catch (Exception e) { - log.error("Exception during processing edge event", e); + log.error("[{}] Exception during processing edge event", tenantId, e); return Futures.immediateFailedFuture(e); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/entityview/BaseEntityViewProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/entityview/BaseEntityViewProcessor.java index 509b2e5196..cca0057707 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/entityview/BaseEntityViewProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/entityview/BaseEntityViewProcessor.java @@ -49,8 +49,8 @@ public abstract class BaseEntityViewProcessor extends BaseEdgeProcessor { EntityView entityViewByName = entityViewService.findEntityViewByTenantIdAndName(tenantId, entityViewName); if (entityViewByName != null && !entityViewByName.getId().equals(entityViewId)) { entityViewName = entityViewName + "_" + StringUtils.randomAlphanumeric(15); - log.warn("Entity view with name {} already exists. Renaming entity view name to {}", - entityViewUpdateMsg.getName(), entityViewName); + log.warn("[{}] Entity view with name {} already exists. Renaming entity view name to {}", + tenantId, entityViewUpdateMsg.getName(), entityViewName); entityViewNameUpdated = true; } entityView.setName(entityViewName); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/BaseRelationProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/BaseRelationProcessor.java index 9038cc9c33..f7be0cf1f4 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/BaseRelationProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/BaseRelationProcessor.java @@ -59,7 +59,7 @@ public abstract class BaseRelationProcessor extends BaseEdgeProcessor { relationService.saveRelation(tenantId, entityRelation); break; } else { - log.warn("Skipping relating update msg because from/to entity doesn't exists on edge, {}", relationUpdateMsg); + log.warn("[{}] Skipping relating update msg because from/to entity doesn't exists on edge, {}", tenantId, relationUpdateMsg); break; } case ENTITY_DELETED_RPC_MESSAGE: diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java index 5942628bf3..c2d172ede0 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java @@ -130,7 +130,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { result.add(processAttributeDeleteMsg(tenantId, entityId, entityData.getAttributeDeleteMsg(), entityData.getEntityType())); } } else { - log.warn("Skipping telemetry update msg because entity doesn't exists on edge, {}", entityData); + log.warn("[{}] Skipping telemetry update msg because entity doesn't exists on edge, {}", tenantId, entityData); } return result; } @@ -172,7 +172,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { } break; default: - log.debug("Using empty metadata for entityId [{}]", entityId); + log.debug("[{}] Using empty metadata for entityId [{}]", tenantId, entityId); break; } return new ImmutablePair<>(metaData, customerId != null ? customerId : new CustomerId(ModelConstants.NULL_UUID)); @@ -193,7 +193,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { @Override public void onFailure(Throwable t) { - log.error("Can't process post telemetry [{}]", msg, t); + log.error("[{}] Can't process post telemetry [{}]", tenantId, msg, t); futureToSet.setException(t); } }); @@ -207,7 +207,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { if (EntityType.DEVICE.equals(entityId.getEntityType())) { DeviceProfile deviceProfile = deviceProfileCache.get(tenantId, new DeviceId(entityId.getId())); if (deviceProfile == null) { - log.warn("[{}] Device profile is null!", entityId); + log.warn("[{}][{}] Device profile is null!", tenantId, entityId); } else { ruleChainId = deviceProfile.getDefaultRuleChainId(); queueName = deviceProfile.getDefaultQueueName(); @@ -215,7 +215,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { } else if (EntityType.ASSET.equals(entityId.getEntityType())) { AssetProfile assetProfile = assetProfileCache.get(tenantId, new AssetId(entityId.getId())); if (assetProfile == null) { - log.warn("[{}] Asset profile is null!", entityId); + log.warn("[{}][{}] Asset profile is null!", tenantId, entityId); } else { ruleChainId = assetProfile.getDefaultRuleChainId(); queueName = assetProfile.getDefaultQueueName(); @@ -237,7 +237,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { @Override public void onFailure(Throwable t) { - log.error("Can't process post attributes [{}]", msg, t); + log.error("[{}] Can't process post attributes [{}]", tenantId, msg, t); futureToSet.setException(t); } }); @@ -267,7 +267,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { @Override public void onFailure(Throwable t) { - log.error("Can't process attributes update [{}]", msg, t); + log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t); futureToSet.setException(t); } }); @@ -275,7 +275,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { @Override public void onFailure(Throwable t) { - log.error("Can't process attributes update [{}]", msg, t); + log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t); futureToSet.setException(t); } }); @@ -300,7 +300,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { @Override public void onFailure(Throwable t) { - log.error("Can't process attribute delete msg [{}]", attributeDeleteMsg, t); + log.error("[{}] Can't process attribute delete msg [{}]", tenantId, attributeDeleteMsg, t); futureToSet.setException(t); } }); @@ -311,7 +311,8 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { }, dbCallbackExecutorService); } - public EntityDataProto convertTelemetryEventToEntityDataProto(EntityType entityType, + public EntityDataProto convertTelemetryEventToEntityDataProto(TenantId tenantId, + EntityType entityType, UUID entityUUID, EdgeEventActionType actionType, JsonNode body) throws JsonProcessingException { @@ -342,11 +343,11 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { entityId = new EdgeId(entityUUID); break; default: - log.warn("Unsupported edge event type [{}]", entityType); + log.warn("[{}] Unsupported edge event type [{}]", tenantId, entityType); return null; } JsonElement entityData = JsonParser.parseString(JacksonUtil.OBJECT_MAPPER.writeValueAsString(body)); - return entityDataMsgConstructor.constructEntityDataMsg(entityId, actionType, entityData); + return entityDataMsgConstructor.constructEntityDataMsg(tenantId, entityId, actionType, entityData); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/TelemetryEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/TelemetryEdgeProcessor.java index 66c5596545..861d766638 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/TelemetryEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/TelemetryEdgeProcessor.java @@ -38,7 +38,8 @@ public class TelemetryEdgeProcessor extends BaseTelemetryProcessor { public DownlinkMsg convertTelemetryEventToDownlink(EdgeEvent edgeEvent) throws JsonProcessingException { EntityType entityType = EntityType.valueOf(edgeEvent.getType().name()); - EntityDataProto entityDataProto = convertTelemetryEventToEntityDataProto(entityType, edgeEvent.getEntityId(), + EntityDataProto entityDataProto = convertTelemetryEventToEntityDataProto( + edgeEvent.getTenantId(), entityType, edgeEvent.getEntityId(), edgeEvent.getAction(), edgeEvent.getBody()); return DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java index 86a97f33db..2483917ce0 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java @@ -177,7 +177,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { entityData.put("kv", attributes); entityData.put("scope", scope); JsonNode body = JacksonUtil.OBJECT_MAPPER.valueToTree(entityData); - log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, body); + log.debug("[{}] Sending attributes data msg, entityId [{}], attributes [{}]", tenantId, entityId, body); future = saveEdgeEvent(tenantId, edge.getId(), entityType, EdgeEventActionType.ATTRIBUTES_UPDATED, entityId, body); } else { future = Futures.immediateFuture(null); @@ -185,7 +185,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { } return Futures.transformAsync(future, v -> processLatestTimeseriesAndAddToEdgeQueue(tenantId, entityId, edge, entityType), dbCallbackExecutorService); } catch (Exception e) { - String errMsg = String.format("[%s] Failed to save attribute updates to the edge [%s]", edge.getId(), attributesRequestMsg); + String errMsg = String.format("[%s][%s] Failed to save attribute updates to the edge [%s]", tenantId, edge.getId(), attributesRequestMsg); log.error(errMsg, e); return Futures.immediateFailedFuture(new RuntimeException(errMsg, e)); } @@ -239,7 +239,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { if (relationsList != null && !relationsList.isEmpty()) { List> futures = new ArrayList<>(); for (List entityRelations : relationsList) { - log.trace("[{}] [{}] [{}] relation(s) are going to be pushed to edge.", edge.getId(), entityId, entityRelations.size()); + log.trace("[{}][{}][{}][{}] relation(s) are going to be pushed to edge.", tenantId, edge.getId(), entityId, entityRelations.size()); for (EntityRelation relation : entityRelations) { try { if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) && @@ -252,7 +252,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { JacksonUtil.OBJECT_MAPPER.valueToTree(relation))); } } catch (Exception e) { - String errMsg = String.format("[%s] Exception during loading relation [%s] to edge on sync!", edge.getId(), relation); + String errMsg = String.format("[%s][%s] Exception during loading relation [%s] to edge on sync!", tenantId, edge.getId(), relation); log.error(errMsg, e); futureToSet.setException(new RuntimeException(errMsg, e)); return; @@ -267,7 +267,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { @Override public void onFailure(Throwable throwable) { - String errMsg = String.format("[%s] Exception during saving edge events [%s]!", edge.getId(), relationRequestMsg); + String errMsg = String.format("[%s][%s] Exception during saving edge events [%s]!", tenantId, edge.getId(), relationRequestMsg); log.error(errMsg, throwable); futureToSet.setException(new RuntimeException(errMsg, throwable)); } @@ -276,7 +276,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { futureToSet.set(null); } } catch (Exception e) { - log.error("Exception during loading relation(s) to edge on sync!", e); + log.error("[{}] Exception during loading relation(s) to edge on sync!", tenantId, e); futureToSet.setException(e); } } @@ -374,7 +374,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { @Override public void onFailure(Throwable t) { - log.error("Exception during loading relation to edge on sync!", t); + log.error("[{}] Exception during loading relation to edge on sync!", tenantId, t); futureToSet.setException(t); } }, dbCallbackExecutorService); diff --git a/application/src/test/java/org/thingsboard/server/edge/AssetProfileEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/AssetProfileEdgeTest.java index f77975f0a9..ebf0af1938 100644 --- a/application/src/test/java/org/thingsboard/server/edge/AssetProfileEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/AssetProfileEdgeTest.java @@ -20,6 +20,7 @@ import com.google.protobuf.AbstractMessage; import com.google.protobuf.ByteString; import org.junit.Assert; import org.junit.Test; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.asset.AssetProfile; import org.thingsboard.server.common.data.id.DashboardId; import org.thingsboard.server.common.data.id.RuleChainId; @@ -30,6 +31,7 @@ import org.thingsboard.server.gen.edge.v1.UplinkMsg; import org.thingsboard.server.gen.edge.v1.UplinkResponseMsg; import java.nio.charset.StandardCharsets; +import java.util.Optional; import java.util.UUID; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @@ -85,7 +87,7 @@ public class AssetProfileEdgeTest extends AbstractEdgeTest { @Test public void testSendAssetProfileToCloud() throws Exception { - RuleChainId ruleChainId = createEdgeRuleChainAndAssignToEdge("Asset Profile Rule Chain"); + RuleChainId edgeRuleChainId = createEdgeRuleChainAndAssignToEdge("Asset Profile Rule Chain"); DashboardId dashboardId = createDashboardAndAssignToEdge("Asset Profile Dashboard"); UUID uuid = Uuids.timeBased(); @@ -96,8 +98,8 @@ public class AssetProfileEdgeTest extends AbstractEdgeTest { assetProfileUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits()); assetProfileUpdateMsgBuilder.setName("Asset Profile On Edge"); assetProfileUpdateMsgBuilder.setDefault(false); - assetProfileUpdateMsgBuilder.setDefaultRuleChainIdMSB(ruleChainId.getId().getMostSignificantBits()); - assetProfileUpdateMsgBuilder.setDefaultRuleChainIdLSB(ruleChainId.getId().getLeastSignificantBits()); + assetProfileUpdateMsgBuilder.setDefaultRuleChainIdMSB(edgeRuleChainId.getId().getMostSignificantBits()); + assetProfileUpdateMsgBuilder.setDefaultRuleChainIdLSB(edgeRuleChainId.getId().getLeastSignificantBits()); assetProfileUpdateMsgBuilder.setDefaultDashboardIdMSB(dashboardId.getId().getMostSignificantBits()); assetProfileUpdateMsgBuilder.setDefaultDashboardIdLSB(dashboardId.getId().getLeastSignificantBits()); assetProfileUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE); @@ -117,6 +119,9 @@ public class AssetProfileEdgeTest extends AbstractEdgeTest { AssetProfile assetProfile = doGet("/api/assetProfile/" + uuid, AssetProfile.class); Assert.assertNotNull(assetProfile); Assert.assertEquals("Asset Profile On Edge", assetProfile.getName()); + Assert.assertEquals(dashboardId, assetProfile.getDefaultDashboardId()); + Assert.assertNull(assetProfile.getDefaultRuleChainId()); + Assert.assertEquals(edgeRuleChainId, assetProfile.getDefaultEdgeRuleChainId()); // delete profile edgeImitator.expectMessageAmount(1); @@ -132,6 +137,47 @@ public class AssetProfileEdgeTest extends AbstractEdgeTest { // cleanup unAssignFromEdgeAndDeleteDashboard(dashboardId); - unAssignFromEdgeAndDeleteRuleChain(ruleChainId); + unAssignFromEdgeAndDeleteRuleChain(edgeRuleChainId); + } + + @Test + public void testSendAssetProfileToCloudWithNameThatAlreadyExistsOnCloud() throws Exception { + String assetProfileOnCloudName = StringUtils.randomAlphanumeric(15); + + edgeImitator.expectMessageAmount(1); + AssetProfile assetProfileOnCloud = this.createAssetProfile(assetProfileOnCloudName); + assetProfileOnCloud = doPost("/api/assetProfile", assetProfileOnCloud, AssetProfile.class); + Assert.assertTrue(edgeImitator.waitForMessages()); + + UUID uuid = Uuids.timeBased(); + + UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder(); + AssetProfileUpdateMsg.Builder assetProfileUpdateMsgBuilder = AssetProfileUpdateMsg.newBuilder(); + assetProfileUpdateMsgBuilder.setIdMSB(uuid.getMostSignificantBits()); + assetProfileUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits()); + assetProfileUpdateMsgBuilder.setName(assetProfileOnCloudName); + assetProfileUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE); + uplinkMsgBuilder.addAssetProfileUpdateMsg(assetProfileUpdateMsgBuilder.build()); + + testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder); + + edgeImitator.expectResponsesAmount(1); + edgeImitator.expectMessageAmount(1); + + edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build()); + + Assert.assertTrue(edgeImitator.waitForResponses()); + Assert.assertTrue(edgeImitator.waitForMessages()); + + Optional assetProfileUpdateMsgOpt = edgeImitator.findMessageByType(AssetProfileUpdateMsg.class); + Assert.assertTrue(assetProfileUpdateMsgOpt.isPresent()); + AssetProfileUpdateMsg latestAssetProfileUpdateMsg = assetProfileUpdateMsgOpt.get(); + Assert.assertNotEquals(assetProfileOnCloudName, latestAssetProfileUpdateMsg.getName()); + + Assert.assertNotEquals(assetProfileOnCloud.getUuidId(), uuid); + + AssetProfile assetProfile = doGet("/api/assetProfile/" + uuid, AssetProfile.class); + Assert.assertNotNull(assetProfile); + Assert.assertNotEquals(assetProfileOnCloudName, assetProfile.getName()); } } 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 fd7dfbaafa..2d21d30174 100644 --- a/application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java @@ -33,7 +33,6 @@ import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.gen.edge.v1.DashboardUpdateMsg; import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.edge.v1.UplinkMsg; -import org.thingsboard.server.gen.edge.v1.UplinkResponseMsg; import java.util.List; import java.util.Optional; @@ -45,12 +44,18 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers. @DaoSqlTest public class DashboardEdgeTest extends AbstractEdgeTest { + private static final int MOBILE_ORDER = 5; + private static final String IMAGE = "data:image/png;base64,iVBORw0KGgoA"; + @Test public void testDashboards() throws Exception { // create dashboard and assign to edge edgeImitator.expectMessageAmount(1); Dashboard dashboard = new Dashboard(); dashboard.setTitle("Edge Test Dashboard"); + dashboard.setMobileHide(true); + dashboard.setImage(IMAGE); + dashboard.setMobileOrder(MOBILE_ORDER); Dashboard savedDashboard = doPost("/api/dashboard", dashboard, Dashboard.class); doPost("/api/edge/" + edge.getUuidId() + "/dashboard/" + savedDashboard.getUuidId(), Dashboard.class); @@ -62,6 +67,9 @@ public class DashboardEdgeTest extends AbstractEdgeTest { Assert.assertEquals(savedDashboard.getUuidId().getMostSignificantBits(), dashboardUpdateMsg.getIdMSB()); Assert.assertEquals(savedDashboard.getUuidId().getLeastSignificantBits(), dashboardUpdateMsg.getIdLSB()); Assert.assertEquals(savedDashboard.getTitle(), dashboardUpdateMsg.getTitle()); + Assert.assertTrue(dashboardUpdateMsg.getMobileHide()); + Assert.assertEquals(IMAGE, dashboardUpdateMsg.getImage()); + Assert.assertEquals(MOBILE_ORDER, dashboardUpdateMsg.getMobileOrder()); testAutoGeneratedCodeByProtobuf(dashboardUpdateMsg); // update dashboard diff --git a/application/src/test/java/org/thingsboard/server/edge/DeviceProfileEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/DeviceProfileEdgeTest.java index d43cca9d8c..9d48824bed 100644 --- a/application/src/test/java/org/thingsboard/server/edge/DeviceProfileEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/DeviceProfileEdgeTest.java @@ -25,7 +25,7 @@ import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceProfileType; import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.OtaPackageInfo; -import org.thingsboard.server.common.data.asset.AssetProfile; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.device.data.PowerMode; import org.thingsboard.server.common.data.device.data.PowerSavingConfiguration; import org.thingsboard.server.common.data.device.profile.CoapDeviceProfileTransportConfiguration; @@ -303,7 +303,7 @@ public class DeviceProfileEdgeTest extends AbstractEdgeTest { UplinkResponseMsg latestResponseMsg = edgeImitator.getLatestResponseMsg(); Assert.assertTrue(latestResponseMsg.getSuccess()); - AssetProfile deviceProfile = doGet("/api/deviceProfile/" + uuid, AssetProfile.class); + DeviceProfile deviceProfile = doGet("/api/deviceProfile/" + uuid, DeviceProfile.class); Assert.assertNotNull(deviceProfile); Assert.assertEquals("Device Profile On Edge", deviceProfile.getName()); @@ -412,4 +412,47 @@ public class DeviceProfileEdgeTest extends AbstractEdgeTest { transportConfiguration.setCoapDeviceTypeConfiguration(coapDeviceTypeConfiguration); return transportConfiguration; } + + @Test + public void testSendDeviceProfileToCloudWithNameThatAlreadyExistsOnCloud() throws Exception { + String deviceProfileOnCloudName = StringUtils.randomAlphanumeric(15); + + edgeImitator.expectMessageAmount(1); + DeviceProfile deviceProfileOnCloud = this.createDeviceProfile(deviceProfileOnCloudName); + deviceProfileOnCloud = doPost("/api/deviceProfile", deviceProfileOnCloud, DeviceProfile.class); + Assert.assertTrue(edgeImitator.waitForMessages()); + + UUID uuid = Uuids.timeBased(); + + UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder(); + DeviceProfileUpdateMsg.Builder deviceProfileUpdateMsgBuilder = DeviceProfileUpdateMsg.newBuilder(); + deviceProfileUpdateMsgBuilder.setIdMSB(uuid.getMostSignificantBits()); + deviceProfileUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits()); + deviceProfileUpdateMsgBuilder.setName(deviceProfileOnCloudName); + deviceProfileUpdateMsgBuilder.setType(DeviceProfileType.DEFAULT.name()); + deviceProfileUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE); + deviceProfileUpdateMsgBuilder.setProfileDataBytes(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfileOnCloud.getProfileData()))); + uplinkMsgBuilder.addDeviceProfileUpdateMsg(deviceProfileUpdateMsgBuilder.build()); + + testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder); + + edgeImitator.expectResponsesAmount(1); + edgeImitator.expectMessageAmount(1); + + edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build()); + + Assert.assertTrue(edgeImitator.waitForResponses()); + Assert.assertTrue(edgeImitator.waitForMessages()); + + Optional deviceProfileUpdateMsgOpt = edgeImitator.findMessageByType(DeviceProfileUpdateMsg.class); + Assert.assertTrue(deviceProfileUpdateMsgOpt.isPresent()); + DeviceProfileUpdateMsg latestDeviceProfileUpdateMsg = deviceProfileUpdateMsgOpt.get(); + Assert.assertNotEquals(deviceProfileOnCloudName, latestDeviceProfileUpdateMsg.getName()); + + Assert.assertNotEquals(deviceProfileOnCloud.getUuidId(), uuid); + + DeviceProfile deviceProfile = doGet("/api/deviceProfile/" + uuid, DeviceProfile.class); + Assert.assertNotNull(deviceProfile); + Assert.assertNotEquals(deviceProfileOnCloudName, deviceProfile.getName()); + } } diff --git a/application/src/test/java/org/thingsboard/server/edge/WidgetEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/WidgetEdgeTest.java index c4fe89a209..54d85a1cef 100644 --- a/application/src/test/java/org/thingsboard/server/edge/WidgetEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/WidgetEdgeTest.java @@ -58,6 +58,7 @@ public class WidgetEdgeTest extends AbstractEdgeTest { ObjectNode descriptor = JacksonUtil.newObjectNode(); descriptor.put("key", "value"); widgetType.setDescriptor(descriptor); + widgetType.setDeprecated(true); WidgetType savedWidgetType = doPost("/api/widgetType", widgetType, WidgetType.class); Assert.assertTrue(edgeImitator.waitForMessages()); latestMessage = edgeImitator.getLatestMessage(); @@ -68,6 +69,7 @@ public class WidgetEdgeTest extends AbstractEdgeTest { Assert.assertEquals(savedWidgetType.getUuidId().getLeastSignificantBits(), widgetTypeUpdateMsg.getIdLSB()); Assert.assertEquals(savedWidgetType.getFqn(), widgetTypeUpdateMsg.getFqn()); Assert.assertEquals(savedWidgetType.getName(), widgetTypeUpdateMsg.getName()); + Assert.assertTrue(widgetTypeUpdateMsg.getDeprecated()); Assert.assertEquals(JacksonUtil.toJsonNode(widgetTypeUpdateMsg.getDescriptorJson()), savedWidgetType.getDescriptor()); // update widget bundle