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

Improve logging

Browse Source
This commit is contained in:
Volodymyr Babak 2020-10-22 14:27:34 +03:00
parent 173eda71ed
commit cee4e5e8d8
5 changed files with 74 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -40,6 +40,7 @@ import org.thingsboard.server.service.edge.EdgeContextComponent;
import org.thingsboard.server.service.state.DefaultDeviceStateService; import org.thingsboard.server.service.state.DefaultDeviceStateService;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import javax.annotation.Nullable;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy; import javax.annotation.PreDestroy;
import java.io.File; import java.io.File;
@ -127,7 +128,10 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
public void updateEdge(Edge edge) { public void updateEdge(Edge edge) {
EdgeGrpcSession session = sessions.get(edge.getId()); EdgeGrpcSession session = sessions.get(edge.getId());
if (session != null && session.isConnected()) { if (session != null && session.isConnected()) {
log.debug("[{}] Updating configuration for edge [{}] [{}]", edge.getTenantId(), edge.getName(), edge.getId());
session.onConfigurationUpdate(edge); session.onConfigurationUpdate(edge);
} else {
log.warn("[{}] Session doesn't exist for edge [{}] [{}]", edge.getTenantId(), edge.getName(), edge.getId());
} }
} }
@ -135,12 +139,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
public void deleteEdge(EdgeId edgeId) { public void deleteEdge(EdgeId edgeId) {
EdgeGrpcSession session = sessions.get(edgeId); EdgeGrpcSession session = sessions.get(edgeId);
if (session != null && session.isConnected()) { if (session != null && session.isConnected()) {
log.debug("Closing and removing session for edge [{}]", edgeId);
session.close(); session.close();
sessions.remove(edgeId); sessions.remove(edgeId);
} }
} }
private void onEdgeConnect(EdgeId edgeId, EdgeGrpcSession edgeGrpcSession) { private void onEdgeConnect(EdgeId edgeId, EdgeGrpcSession edgeGrpcSession) {
log.debug("[{}] onEdgeConnect [{}]", edgeId, edgeGrpcSession.getSessionId());
sessions.put(edgeId, edgeGrpcSession); sessions.put(edgeId, edgeGrpcSession);
save(edgeId, DefaultDeviceStateService.ACTIVITY_STATE, true); save(edgeId, DefaultDeviceStateService.ACTIVITY_STATE, true);
save(edgeId, DefaultDeviceStateService.LAST_CONNECT_TIME, System.currentTimeMillis()); save(edgeId, DefaultDeviceStateService.LAST_CONNECT_TIME, System.currentTimeMillis());
@ -180,12 +186,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
} }
private void onEdgeDisconnect(EdgeId edgeId) { private void onEdgeDisconnect(EdgeId edgeId) {
log.debug("[{}] onEdgeDisconnect", edgeId);
sessions.remove(edgeId); sessions.remove(edgeId);
save(edgeId, DefaultDeviceStateService.ACTIVITY_STATE, false); save(edgeId, DefaultDeviceStateService.ACTIVITY_STATE, false);
save(edgeId, DefaultDeviceStateService.LAST_DISCONNECT_TIME, System.currentTimeMillis()); save(edgeId, DefaultDeviceStateService.LAST_DISCONNECT_TIME, System.currentTimeMillis());
} }
private void save(EdgeId edgeId, String key, long value) { private void save(EdgeId edgeId, String key, long value) {
log.debug("[{}] Updating long edge telemetry [{}] [{}]", edgeId, key, value);
if (persistToTelemetry) { if (persistToTelemetry) {
tsSubService.saveAndNotify( tsSubService.saveAndNotify(
TenantId.SYS_TENANT_ID, edgeId, TenantId.SYS_TENANT_ID, edgeId,
@ -197,6 +205,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
} }
private void save(EdgeId edgeId, String key, boolean value) { private void save(EdgeId edgeId, String key, boolean value) {
log.debug("[{}] Updating boolean edge telemetry [{}] [{}]", edgeId, key, value);
if (persistToTelemetry) { if (persistToTelemetry) {
tsSubService.saveAndNotify( tsSubService.saveAndNotify(
TenantId.SYS_TENANT_ID, edgeId, TenantId.SYS_TENANT_ID, edgeId,
@ -219,7 +228,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
} }
@Override @Override
public void onSuccess(@javax.annotation.Nullable Void result) { public void onSuccess(@Nullable Void result) {
log.trace("[{}] Successfully updated attribute [{}] with value [{}]", edgeId, key, value); log.trace("[{}] Successfully updated attribute [{}] with value [{}]", edgeId, key, value);
} }

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

@ -223,26 +223,29 @@ public final class EdgeGrpcSession implements Closeable {
} }
latch.countDown(); latch.countDown();
} catch (Exception e) { } catch (Exception e) {
log.error("Can't process downlink response message [{}]", msg, e); log.error("[{}] Can't process downlink response message [{}]", this.sessionId, msg, e);
} }
} }
private void sendResponseMsg(ResponseMsg responseMsg) { private void sendResponseMsg(ResponseMsg responseMsg) {
log.trace("[{}] Sending response msg [{}]", this.sessionId, responseMsg);
if (isConnected()) { if (isConnected()) {
try { try {
responseMsgLock.lock(); responseMsgLock.lock();
outputStream.onNext(responseMsg); outputStream.onNext(responseMsg);
} catch (Exception e) { } catch (Exception e) {
log.error("Failed to send response message [{}]", responseMsg, e); log.error("[{}] Failed to send response message [{}]", this.sessionId, responseMsg, e);
connected = false; connected = false;
sessionCloseListener.accept(edge.getId()); sessionCloseListener.accept(edge.getId());
} finally { } finally {
responseMsgLock.unlock(); responseMsgLock.unlock();
} }
log.trace("[{}] Response msg successfully sent [{}]", this.sessionId, responseMsg);
} }
} }
void onConfigurationUpdate(Edge edge) { void onConfigurationUpdate(Edge edge) {
log.debug("[{}] onConfigurationUpdate [{}]", this.sessionId, edge);
try { try {
this.edge = edge; this.edge = edge;
EdgeUpdateMsg edgeConfig = EdgeUpdateMsg.newBuilder() EdgeUpdateMsg edgeConfig = EdgeUpdateMsg.newBuilder()
@ -251,13 +254,15 @@ public final class EdgeGrpcSession implements Closeable {
.setEdgeUpdateMsg(edgeConfig) .setEdgeUpdateMsg(edgeConfig)
.build()); .build());
} catch (Exception e) { } catch (Exception e) {
log.error("Failed to construct proto objects!", e); log.error("[{}] Failed to construct proto objects!", this.sessionId, e);
} }
} }
void processHandleMessages() throws ExecutionException, InterruptedException { void processHandleMessages() throws ExecutionException, InterruptedException {
log.trace("[{}] processHandleMessages started", this.sessionId);
if (isConnected()) { if (isConnected()) {
Long queueStartTs = getQueueStartTs().get(); Long queueStartTs = getQueueStartTs().get();
log.trace("[{}] trying to find edge events using queue start ts [{}]", this.sessionId, queueStartTs);
TimePageLink pageLink = new TimePageLink(ctx.getEdgeEventStorageSettings().getMaxReadRecordsCount(), queueStartTs, null, true); TimePageLink pageLink = new TimePageLink(ctx.getEdgeEventStorageSettings().getMaxReadRecordsCount(), queueStartTs, null, true);
TimePageData<EdgeEvent> pageData; TimePageData<EdgeEvent> pageData;
UUID ifOffset = null; UUID ifOffset = null;
@ -267,7 +272,7 @@ public final class EdgeGrpcSession implements Closeable {
if (isConnected() && !pageData.getData().isEmpty()) { 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.sessionId, pageData.getData().size());
List<DownlinkMsg> downlinkMsgsPack = convertToDownlinkMsgsPack(pageData.getData()); List<DownlinkMsg> downlinkMsgsPack = convertToDownlinkMsgsPack(pageData.getData());
log.trace("[{}] downlink msg(s) are going to be send.", downlinkMsgsPack.size()); log.trace("[{}] [{}] downlink msg(s) are going to be send.", this.sessionId, downlinkMsgsPack.size());
latch = new CountDownLatch(downlinkMsgsPack.size()); latch = new CountDownLatch(downlinkMsgsPack.size());
for (DownlinkMsg downlinkMsg : downlinkMsgsPack) { for (DownlinkMsg downlinkMsg : downlinkMsgsPack) {
@ -280,14 +285,14 @@ public final class EdgeGrpcSession implements Closeable {
success = latch.await(10, TimeUnit.SECONDS); success = latch.await(10, TimeUnit.SECONDS);
if (!success) { if (!success) {
log.warn("Failed to deliver the batch: {}", downlinkMsgsPack); log.warn("[{}] Failed to deliver the batch: {}", this.sessionId, downlinkMsgsPack);
} }
} }
if (isConnected() && (!success || pageData.hasNext())) { if (isConnected() && (!success || pageData.hasNext())) {
try { try {
Thread.sleep(ctx.getEdgeEventStorageSettings().getSleepIntervalBetweenBatches()); Thread.sleep(ctx.getEdgeEventStorageSettings().getSleepIntervalBetweenBatches());
} catch (InterruptedException e) { } catch (InterruptedException e) {
log.error("Error during sleep between batches", e); log.error("[{}] Error during sleep between batches", this.sessionId, e);
} }
if (success) { if (success) {
pageLink = pageData.getNextPageLink(); pageLink = pageData.getNextPageLink();
@ -302,15 +307,16 @@ public final class EdgeGrpcSession implements Closeable {
try { try {
Thread.sleep(ctx.getEdgeEventStorageSettings().getNoRecordsSleepInterval()); Thread.sleep(ctx.getEdgeEventStorageSettings().getNoRecordsSleepInterval());
} catch (InterruptedException e) { } catch (InterruptedException e) {
log.error("Error during sleep", e); log.error("[{}] Error during sleep between no records interval", this.sessionId, e);
} }
} }
log.trace("[{}] processHandleMessages finished", this.sessionId);
} }
private List<DownlinkMsg> convertToDownlinkMsgsPack(List<EdgeEvent> edgeEvents) { private List<DownlinkMsg> convertToDownlinkMsgsPack(List<EdgeEvent> edgeEvents) {
List<DownlinkMsg> result = new ArrayList<>(); List<DownlinkMsg> result = new ArrayList<>();
for (EdgeEvent edgeEvent : edgeEvents) { for (EdgeEvent edgeEvent : edgeEvents) {
log.trace("Processing edge event [{}]", edgeEvent); log.trace("[{}] Processing edge event [{}]", this.sessionId, edgeEvent);
try { try {
DownlinkMsg downlinkMsg = null; DownlinkMsg downlinkMsg = null;
switch (edgeEvent.getAction()) { switch (edgeEvent.getAction()) {
@ -406,13 +412,14 @@ public final class EdgeGrpcSession implements Closeable {
} }
private void updateQueueStartTs(Long newStartTs) { private void updateQueueStartTs(Long newStartTs) {
log.trace("[{}] updating QueueStartTs [{}][{}]", this.sessionId, edge.getId(), newStartTs);
newStartTs = ++newStartTs; // increments ts by 1 - next edge event search starts from current offset + 1 newStartTs = ++newStartTs; // increments ts by 1 - next edge event search starts from current offset + 1
List<AttributeKvEntry> attributes = Collections.singletonList(new BaseAttributeKvEntry(new LongDataEntry(QUEUE_START_TS_ATTR_KEY, newStartTs), System.currentTimeMillis())); List<AttributeKvEntry> attributes = Collections.singletonList(new BaseAttributeKvEntry(new LongDataEntry(QUEUE_START_TS_ATTR_KEY, newStartTs), System.currentTimeMillis()));
ctx.getAttributesService().save(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, attributes); ctx.getAttributesService().save(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, attributes);
} }
private DownlinkMsg processTelemetryMessage(EdgeEvent edgeEvent) { private DownlinkMsg processTelemetryMessage(EdgeEvent edgeEvent) {
log.trace("Executing processTelemetryMessage, edgeEvent [{}]", edgeEvent); log.trace("[{}] Executing processTelemetryMessage, edgeEvent [{}]", this.sessionId, edgeEvent);
EntityId entityId = null; EntityId entityId = null;
switch (edgeEvent.getType()) { switch (edgeEvent.getType()) {
case DEVICE: case DEVICE:
@ -436,11 +443,11 @@ public final class EdgeGrpcSession implements Closeable {
} }
DownlinkMsg downlinkMsg = null; DownlinkMsg downlinkMsg = null;
if (entityId != null) { if (entityId != null) {
log.debug("Sending telemetry data msg, entityId [{}], body [{}]", edgeEvent.getEntityId(), edgeEvent.getBody()); log.debug("[{}] Sending telemetry data msg, entityId [{}], body [{}]", this.sessionId, edgeEvent.getEntityId(), edgeEvent.getBody());
try { try {
downlinkMsg = constructEntityDataProtoMsg(entityId, edgeEvent.getAction(), JsonUtils.parse(mapper.writeValueAsString(edgeEvent.getBody()))); downlinkMsg = constructEntityDataProtoMsg(entityId, edgeEvent.getAction(), JsonUtils.parse(mapper.writeValueAsString(edgeEvent.getBody())));
} catch (Exception e) { } catch (Exception e) {
log.warn("Can't send telemetry data msg, entityId [{}], body [{}]", edgeEvent.getEntityId(), edgeEvent.getBody(), e); log.warn("[{}] Can't send telemetry data msg, entityId [{}], body [{}]", this.sessionId, edgeEvent.getEntityId(), edgeEvent.getBody(), e);
} }
} }
return downlinkMsg; return downlinkMsg;
@ -450,9 +457,6 @@ public final class EdgeGrpcSession implements Closeable {
UpdateMsgType msgType = getResponseMsgType(edgeEvent.getAction()); UpdateMsgType msgType = getResponseMsgType(edgeEvent.getAction());
log.trace("Executing processEntityMessage, edgeEvent [{}], action [{}], msgType [{}]", edgeEvent, action, msgType); log.trace("Executing processEntityMessage, edgeEvent [{}], action [{}], msgType [{}]", edgeEvent, action, msgType);
switch (edgeEvent.getType()) { switch (edgeEvent.getType()) {
case EDGE:
// TODO: voba - add edge update logic
return null;
case DEVICE: case DEVICE:
return processDevice(edgeEvent, msgType, action); return processDevice(edgeEvent, msgType, action);
case ASSET: case ASSET:
@ -523,6 +527,7 @@ public final class EdgeGrpcSession implements Closeable {
} }
break; break;
} }
log.trace("[{}] device processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -554,6 +559,7 @@ public final class EdgeGrpcSession implements Closeable {
.build(); .build();
break; break;
} }
log.trace("[{}] asset processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -585,6 +591,7 @@ public final class EdgeGrpcSession implements Closeable {
.build(); .build();
break; break;
} }
log.trace("[{}] entity view processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -619,6 +626,7 @@ public final class EdgeGrpcSession implements Closeable {
.build(); .build();
break; break;
} }
log.trace("[{}] dashboard processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -645,6 +653,7 @@ public final class EdgeGrpcSession implements Closeable {
.build(); .build();
break; break;
} }
log.trace("[{}] customer processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -671,6 +680,7 @@ public final class EdgeGrpcSession implements Closeable {
.build(); .build();
break; break;
} }
log.trace("[{}] rule chain processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -688,6 +698,7 @@ public final class EdgeGrpcSession implements Closeable {
.build(); .build();
} }
} }
log.trace("[{}] rule chain metadata processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -720,6 +731,7 @@ public final class EdgeGrpcSession implements Closeable {
.build(); .build();
} }
} }
log.trace("[{}] user processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -734,9 +746,11 @@ public final class EdgeGrpcSession implements Closeable {
private DownlinkMsg processRelation(EdgeEvent edgeEvent, UpdateMsgType msgType) { private DownlinkMsg processRelation(EdgeEvent edgeEvent, UpdateMsgType msgType) {
EntityRelation entityRelation = mapper.convertValue(edgeEvent.getBody(), EntityRelation.class); EntityRelation entityRelation = mapper.convertValue(edgeEvent.getBody(), EntityRelation.class);
RelationUpdateMsg r = ctx.getRelationMsgConstructor().constructRelationUpdatedMsg(msgType, entityRelation); RelationUpdateMsg r = ctx.getRelationMsgConstructor().constructRelationUpdatedMsg(msgType, entityRelation);
return DownlinkMsg.newBuilder() DownlinkMsg downlinkMsg = DownlinkMsg.newBuilder()
.addAllRelationUpdateMsg(Collections.singletonList(r)) .addAllRelationUpdateMsg(Collections.singletonList(r))
.build(); .build();
log.trace("[{}] relation processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg;
} }
private DownlinkMsg processAlarm(EdgeEvent edgeEvent, UpdateMsgType msgType) { private DownlinkMsg processAlarm(EdgeEvent edgeEvent, UpdateMsgType msgType) {
@ -752,6 +766,7 @@ public final class EdgeGrpcSession implements Closeable {
} catch (Exception e) { } catch (Exception e) {
log.error("Can't process alarm msg [{}] [{}]", edgeEvent, msgType, e); log.error("Can't process alarm msg [{}] [{}]", edgeEvent, msgType, e);
} }
log.trace("[{}] alarm processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -778,6 +793,7 @@ public final class EdgeGrpcSession implements Closeable {
.build(); .build();
break; break;
} }
log.trace("[{}] widget bundle processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
@ -804,15 +820,18 @@ public final class EdgeGrpcSession implements Closeable {
.build(); .build();
break; break;
} }
log.trace("[{}] widget type processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg; return downlinkMsg;
} }
private DownlinkMsg processAdminSettings(EdgeEvent edgeEvent) { private DownlinkMsg processAdminSettings(EdgeEvent edgeEvent) {
AdminSettings adminSettings = mapper.convertValue(edgeEvent.getBody(), AdminSettings.class); AdminSettings adminSettings = mapper.convertValue(edgeEvent.getBody(), AdminSettings.class);
AdminSettingsUpdateMsg t = ctx.getAdminSettingsMsgConstructor().constructAdminSettingsUpdateMsg(adminSettings); AdminSettingsUpdateMsg t = ctx.getAdminSettingsMsgConstructor().constructAdminSettingsUpdateMsg(adminSettings);
return DownlinkMsg.newBuilder() DownlinkMsg downlinkMsg = DownlinkMsg.newBuilder()
.addAllAdminSettingsUpdateMsg(Collections.singletonList(t)) .addAllAdminSettingsUpdateMsg(Collections.singletonList(t))
.build(); .build();
log.trace("[{}] admin settings processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg;
} }
private UpdateMsgType getResponseMsgType(EdgeEventActionType actionType) { private UpdateMsgType getResponseMsgType(EdgeEventActionType actionType) {
@ -841,9 +860,11 @@ public final class EdgeGrpcSession implements Closeable {
private DownlinkMsg constructEntityDataProtoMsg(EntityId entityId, EdgeEventActionType actionType, JsonElement entityData) { private DownlinkMsg constructEntityDataProtoMsg(EntityId entityId, EdgeEventActionType actionType, JsonElement entityData) {
EntityDataProto entityDataProto = ctx.getEntityDataMsgConstructor().constructEntityDataMsg(entityId, actionType, entityData); EntityDataProto entityDataProto = ctx.getEntityDataMsgConstructor().constructEntityDataMsg(entityId, actionType, entityData);
DownlinkMsg.Builder builder = DownlinkMsg.newBuilder() DownlinkMsg downlinkMsg = DownlinkMsg.newBuilder()
.addAllEntityData(Collections.singletonList(entityDataProto)); .addAllEntityData(Collections.singletonList(entityDataProto))
return builder.build(); .build();
log.trace("[{}] entity data proto processed [{}]", this.sessionId, downlinkMsg);
return downlinkMsg;
} }
private ListenableFuture<List<Void>> processUplinkMsg(UplinkMsg uplinkMsg) { private ListenableFuture<List<Void>> processUplinkMsg(UplinkMsg uplinkMsg) {
@ -854,7 +875,6 @@ public final class EdgeGrpcSession implements Closeable {
result.addAll(ctx.getTelemetryProcessor().onTelemetryUpdate(edge.getTenantId(), entityData)); result.addAll(ctx.getTelemetryProcessor().onTelemetryUpdate(edge.getTenantId(), entityData));
} }
} }
if (uplinkMsg.getDeviceUpdateMsgList() != null && !uplinkMsg.getDeviceUpdateMsgList().isEmpty()) { if (uplinkMsg.getDeviceUpdateMsgList() != null && !uplinkMsg.getDeviceUpdateMsgList().isEmpty()) {
for (DeviceUpdateMsg deviceUpdateMsg : uplinkMsg.getDeviceUpdateMsgList()) { for (DeviceUpdateMsg deviceUpdateMsg : uplinkMsg.getDeviceUpdateMsgList()) {
result.add(ctx.getDeviceProcessor().onDeviceUpdate(edge.getTenantId(), edge, deviceUpdateMsg)); result.add(ctx.getDeviceProcessor().onDeviceUpdate(edge.getTenantId(), edge, deviceUpdateMsg));
@ -906,12 +926,13 @@ public final class EdgeGrpcSession implements Closeable {
} }
} }
} catch (Exception e) { } catch (Exception e) {
log.error("Can't process uplink msg [{}]", uplinkMsg, e); log.error("[{}] Can't process uplink msg [{}]", this.sessionId, uplinkMsg, e);
} }
return Futures.allAsList(result); return Futures.allAsList(result);
} }
private ConnectResponseMsg processConnect(ConnectRequestMsg request) { private ConnectResponseMsg processConnect(ConnectRequestMsg request) {
log.trace("[{}] processConnect [{}]", this.sessionId, request);
Optional<Edge> optional = ctx.getEdgeService().findEdgeByRoutingKey(TenantId.SYS_TENANT_ID, request.getEdgeRoutingKey()); Optional<Edge> optional = ctx.getEdgeService().findEdgeByRoutingKey(TenantId.SYS_TENANT_ID, request.getEdgeRoutingKey());
if (optional.isPresent()) { if (optional.isPresent()) {
edge = optional.get(); edge = optional.get();
@ -941,7 +962,7 @@ public final class EdgeGrpcSession implements Closeable {
.setConfiguration(EdgeConfiguration.getDefaultInstance()).build(); .setConfiguration(EdgeConfiguration.getDefaultInstance()).build();
} }
private EdgeConfiguration constructEdgeConfigProto(Edge edge) throws JsonProcessingException { private EdgeConfiguration constructEdgeConfigProto(Edge edge) {
return EdgeConfiguration.newBuilder() return EdgeConfiguration.newBuilder()
.setEdgeIdMSB(edge.getId().getId().getMostSignificantBits()) .setEdgeIdMSB(edge.getId().getId().getMostSignificantBits())
.setEdgeIdLSB(edge.getId().getId().getLeastSignificantBits()) .setEdgeIdLSB(edge.getId().getId().getLeastSignificantBits())
@ -958,6 +979,7 @@ public final class EdgeGrpcSession implements Closeable {
@Override @Override
public void close() { public void close() {
log.debug("[{}] Closing session", sessionId);
connected = false; connected = false;
try { try {
outputStream.onCompleted(); outputStream.onCompleted();

16
application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java
View File

@ -140,6 +140,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
@Override @Override
public void sync(Edge edge) { public void sync(Edge edge) {
log.trace("[{}] staring sync process for edge [{}]", edge.getTenantId(), edge.getName());
try { try {
syncWidgetsBundleAndWidgetTypes(edge); syncWidgetsBundleAndWidgetTypes(edge);
syncAdminSettings(edge); syncAdminSettings(edge);
@ -155,6 +156,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
} }
private void syncRuleChains(Edge edge) { private void syncRuleChains(Edge edge) {
log.trace("[{}] syncRuleChains [{}]", edge.getTenantId(), edge.getName());
try { try {
ListenableFuture<TimePageData<RuleChain>> future = ListenableFuture<TimePageData<RuleChain>> future =
ruleChainService.findRuleChainsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE)); ruleChainService.findRuleChainsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
@ -180,6 +182,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
} }
private void syncDevices(Edge edge) { private void syncDevices(Edge edge) {
log.trace("[{}] syncDevices [{}]", edge.getTenantId(), edge.getName());
try { try {
ListenableFuture<TimePageData<Device>> future = ListenableFuture<TimePageData<Device>> future =
deviceService.findDevicesByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE)); deviceService.findDevicesByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
@ -205,6 +208,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
} }
private void syncAssets(Edge edge) { private void syncAssets(Edge edge) {
log.trace("[{}] syncAssets [{}]", edge.getTenantId(), edge.getName());
try { try {
ListenableFuture<TimePageData<Asset>> future = assetService.findAssetsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE)); ListenableFuture<TimePageData<Asset>> future = assetService.findAssetsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
Futures.addCallback(future, new FutureCallback<TimePageData<Asset>>() { Futures.addCallback(future, new FutureCallback<TimePageData<Asset>>() {
@ -229,6 +233,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
} }
private void syncEntityViews(Edge edge) { private void syncEntityViews(Edge edge) {
log.trace("[{}] syncEntityViews [{}]", edge.getTenantId(), edge.getName());
try { try {
ListenableFuture<TimePageData<EntityView>> future = entityViewService.findEntityViewsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE)); ListenableFuture<TimePageData<EntityView>> future = entityViewService.findEntityViewsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
Futures.addCallback(future, new FutureCallback<TimePageData<EntityView>>() { Futures.addCallback(future, new FutureCallback<TimePageData<EntityView>>() {
@ -253,6 +258,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
} }
private void syncDashboards(Edge edge) { private void syncDashboards(Edge edge) {
log.trace("[{}] syncDashboards [{}]", edge.getTenantId(), edge.getName());
try { try {
ListenableFuture<TimePageData<DashboardInfo>> future = dashboardService.findDashboardsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE)); ListenableFuture<TimePageData<DashboardInfo>> future = dashboardService.findDashboardsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
Futures.addCallback(future, new FutureCallback<TimePageData<DashboardInfo>>() { Futures.addCallback(future, new FutureCallback<TimePageData<DashboardInfo>>() {
@ -277,6 +283,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
} }
private void syncUsers(Edge edge) { private void syncUsers(Edge edge) {
log.trace("[{}] syncUsers [{}]", edge.getTenantId(), edge.getName());
try { try {
TextPageData<User> pageData = userService.findTenantAdmins(edge.getTenantId(), new TextPageLink(Integer.MAX_VALUE)); TextPageData<User> pageData = userService.findTenantAdmins(edge.getTenantId(), new TextPageLink(Integer.MAX_VALUE));
pushUsersToEdge(pageData, edge); pushUsersToEdge(pageData, edge);
@ -291,6 +298,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
} }
private void syncWidgetsBundleAndWidgetTypes(Edge edge) { private void syncWidgetsBundleAndWidgetTypes(Edge edge) {
log.trace("[{}] syncWidgetsBundleAndWidgetTypes [{}]", edge.getTenantId(), edge.getName());
List<WidgetsBundle> widgetsBundlesToPush = new ArrayList<>(); List<WidgetsBundle> widgetsBundlesToPush = new ArrayList<>();
List<WidgetType> widgetTypesToPush = new ArrayList<>(); List<WidgetType> widgetTypesToPush = new ArrayList<>();
widgetsBundlesToPush.addAll(widgetsBundleService.findAllTenantWidgetsBundlesByTenantId(edge.getTenantId())); widgetsBundlesToPush.addAll(widgetsBundleService.findAllTenantWidgetsBundlesByTenantId(edge.getTenantId()));
@ -309,6 +317,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
} }
private void syncAdminSettings(Edge edge) { private void syncAdminSettings(Edge edge) {
log.trace("[{}] syncAdminSettings [{}]", edge.getTenantId(), edge.getName());
try { try {
AdminSettings systemMailSettings = adminSettingsService.findAdminSettingsByKey(TenantId.SYS_TENANT_ID, "mail"); AdminSettings systemMailSettings = adminSettingsService.findAdminSettingsByKey(TenantId.SYS_TENANT_ID, "mail");
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(systemMailSettings)); saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(systemMailSettings));
@ -386,6 +395,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
@Override @Override
public ListenableFuture<Void> processRuleChainMetadataRequestMsg(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg) { public ListenableFuture<Void> processRuleChainMetadataRequestMsg(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg) {
log.trace("[{}] processRuleChainMetadataRequestMsg [{}][{}]", edge.getTenantId(), edge.getName(), ruleChainMetadataRequestMsg);
SettableFuture<Void> futureToSet = SettableFuture.create(); SettableFuture<Void> futureToSet = SettableFuture.create();
if (ruleChainMetadataRequestMsg.getRuleChainIdMSB() != 0 && ruleChainMetadataRequestMsg.getRuleChainIdLSB() != 0) { if (ruleChainMetadataRequestMsg.getRuleChainIdMSB() != 0 && ruleChainMetadataRequestMsg.getRuleChainIdLSB() != 0) {
RuleChainId ruleChainId = RuleChainId ruleChainId =
@ -409,6 +419,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
@Override @Override
public ListenableFuture<Void> processAttributesRequestMsg(Edge edge, AttributesRequestMsg attributesRequestMsg) { public ListenableFuture<Void> processAttributesRequestMsg(Edge edge, AttributesRequestMsg attributesRequestMsg) {
log.trace("[{}] processAttributesRequestMsg [{}][{}]", edge.getTenantId(), edge.getName(), attributesRequestMsg);
EntityId entityId = EntityIdFactory.getByTypeAndUuid( EntityId entityId = EntityIdFactory.getByTypeAndUuid(
EntityType.valueOf(attributesRequestMsg.getEntityType()), EntityType.valueOf(attributesRequestMsg.getEntityType()),
new UUID(attributesRequestMsg.getEntityIdMSB(), attributesRequestMsg.getEntityIdLSB())); new UUID(attributesRequestMsg.getEntityIdMSB(), attributesRequestMsg.getEntityIdLSB()));
@ -482,6 +493,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
@Override @Override
public ListenableFuture<Void> processRelationRequestMsg(Edge edge, RelationRequestMsg relationRequestMsg) { public ListenableFuture<Void> processRelationRequestMsg(Edge edge, RelationRequestMsg relationRequestMsg) {
log.trace("[{}] processRelationRequestMsg [{}][{}]", edge.getTenantId(), edge.getName(), relationRequestMsg);
EntityId entityId = EntityIdFactory.getByTypeAndUuid( EntityId entityId = EntityIdFactory.getByTypeAndUuid(
EntityType.valueOf(relationRequestMsg.getEntityType()), EntityType.valueOf(relationRequestMsg.getEntityType()),
new UUID(relationRequestMsg.getEntityIdMSB(), relationRequestMsg.getEntityIdLSB())); new UUID(relationRequestMsg.getEntityIdMSB(), relationRequestMsg.getEntityIdLSB()));
@ -528,6 +540,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
@Override @Override
public ListenableFuture<Void> processDeviceCredentialsRequestMsg(Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg) { public ListenableFuture<Void> processDeviceCredentialsRequestMsg(Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg) {
log.trace("[{}] processDeviceCredentialsRequestMsg [{}][{}]", edge.getTenantId(), edge.getName(), deviceCredentialsRequestMsg);
SettableFuture<Void> futureToSet = SettableFuture.create(); SettableFuture<Void> futureToSet = SettableFuture.create();
if (deviceCredentialsRequestMsg.getDeviceIdMSB() != 0 && deviceCredentialsRequestMsg.getDeviceIdLSB() != 0) { if (deviceCredentialsRequestMsg.getDeviceIdMSB() != 0 && deviceCredentialsRequestMsg.getDeviceIdLSB() != 0) {
DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsRequestMsg.getDeviceIdMSB(), deviceCredentialsRequestMsg.getDeviceIdLSB())); DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsRequestMsg.getDeviceIdMSB(), deviceCredentialsRequestMsg.getDeviceIdLSB()));
@ -550,6 +563,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
@Override @Override
public ListenableFuture<Void> processUserCredentialsRequestMsg(Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg) { public ListenableFuture<Void> processUserCredentialsRequestMsg(Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg) {
log.trace("[{}] processUserCredentialsRequestMsg [{}][{}]", edge.getTenantId(), edge.getName(), userCredentialsRequestMsg);
SettableFuture<Void> futureToSet = SettableFuture.create(); SettableFuture<Void> futureToSet = SettableFuture.create();
if (userCredentialsRequestMsg.getUserIdMSB() != 0 && userCredentialsRequestMsg.getUserIdLSB() != 0) { if (userCredentialsRequestMsg.getUserIdMSB() != 0 && userCredentialsRequestMsg.getUserIdLSB() != 0) {
UserId userId = new UserId(new UUID(userCredentialsRequestMsg.getUserIdMSB(), userCredentialsRequestMsg.getUserIdLSB())); UserId userId = new UserId(new UUID(userCredentialsRequestMsg.getUserIdMSB(), userCredentialsRequestMsg.getUserIdLSB()));
@ -576,7 +590,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
EdgeEventActionType action, EdgeEventActionType action,
EntityId entityId, EntityId entityId,
JsonNode body) { JsonNode body) {
log.debug("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], type [{}], action[{}], entityId [{}], body [{}]", log.trace("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], type [{}], action[{}], entityId [{}], body [{}]",
tenantId, edgeId, type, action, entityId, body); tenantId, edgeId, type, action, entityId, body);
EdgeEvent edgeEvent = new EdgeEvent(); EdgeEvent edgeEvent = new EdgeEvent();

1
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceProcessor.java
View File

@ -221,6 +221,7 @@ public class DeviceProcessor extends BaseProcessor {
} }
public ListenableFuture<Void> processDeviceRpcCallResponseMsg(TenantId tenantId, DeviceRpcCallMsg deviceRpcCallMsg) { public ListenableFuture<Void> processDeviceRpcCallResponseMsg(TenantId tenantId, DeviceRpcCallMsg deviceRpcCallMsg) {
log.trace("[{}] processDeviceRpcCallResponseMsg [{}]", tenantId, deviceRpcCallMsg);
SettableFuture<Void> futureToSet = SettableFuture.create(); SettableFuture<Void> futureToSet = SettableFuture.create();
UUID uuid = new UUID(deviceRpcCallMsg.getRequestIdMSB(), deviceRpcCallMsg.getRequestIdLSB()); UUID uuid = new UUID(deviceRpcCallMsg.getRequestIdMSB(), deviceRpcCallMsg.getRequestIdLSB());
FromDeviceRpcResponse response; FromDeviceRpcResponse response;

4
dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java
View File

@ -174,6 +174,7 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
@Override @Override
public Edge assignEdgeToCustomer(TenantId tenantId, EdgeId edgeId, CustomerId customerId) { public Edge assignEdgeToCustomer(TenantId tenantId, EdgeId edgeId, CustomerId customerId) {
log.trace("[{}] Executing assignEdgeToCustomer [{}][{}]", tenantId, edgeId, customerId);
Edge edge = findEdgeById(tenantId, edgeId); Edge edge = findEdgeById(tenantId, edgeId);
edge.setCustomerId(customerId); edge.setCustomerId(customerId);
return saveEdge(edge); return saveEdge(edge);
@ -181,6 +182,7 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
@Override @Override
public Edge unassignEdgeFromCustomer(TenantId tenantId, EdgeId edgeId) { public Edge unassignEdgeFromCustomer(TenantId tenantId, EdgeId edgeId) {
log.trace("[{}] Executing unassignEdgeFromCustomer [{}]", tenantId, edgeId);
Edge edge = findEdgeById(tenantId, edgeId); Edge edge = findEdgeById(tenantId, edgeId);
edge.setCustomerId(null); edge.setCustomerId(null);
return saveEdge(edge); return saveEdge(edge);
@ -279,6 +281,7 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
@Override @Override
public ListenableFuture<List<Edge>> findEdgesByQuery(TenantId tenantId, EdgeSearchQuery query) { public ListenableFuture<List<Edge>> findEdgesByQuery(TenantId tenantId, EdgeSearchQuery query) {
log.trace("[{}] Executing findEdgesByQuery [{}]", tenantId, query);
ListenableFuture<List<EntityRelation>> relations = relationService.findByQuery(tenantId, query.toEntitySearchQuery()); ListenableFuture<List<EntityRelation>> relations = relationService.findByQuery(tenantId, query.toEntitySearchQuery());
ListenableFuture<List<Edge>> edges = Futures.transformAsync(relations, r -> { ListenableFuture<List<Edge>> edges = Futures.transformAsync(relations, r -> {
EntitySearchDirection direction = query.toEntitySearchQuery().getParameters().getDirection(); EntitySearchDirection direction = query.toEntitySearchQuery().getParameters().getDirection();
@ -455,6 +458,7 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
@Override @Override
public ListenableFuture<List<EdgeId>> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId) { public ListenableFuture<List<EdgeId>> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId) {
log.trace("[{}] Executing findRelatedEdgeIdsByEntityId [{}]", tenantId, entityId);
if (EntityType.TENANT.equals(entityId.getEntityType())) { if (EntityType.TENANT.equals(entityId.getEntityType())) {
TextPageData<Edge> edgesByTenantId = findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); TextPageData<Edge> edgesByTenantId = findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE));
return Futures.immediateFuture(edgesByTenantId.getData().stream().map(IdBased::getId).collect(Collectors.toList())); return Futures.immediateFuture(edgesByTenantId.getData().stream().map(IdBased::getId).collect(Collectors.toList()));