Browse Source

added logs

pull/12620/head
IrynaMatveieva 2 years ago
parent
commit
ebab88ac6a
  1. 7
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  2. 28
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java

7
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java

@ -83,12 +83,15 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} }
public void process(CalculatedFieldStateRestoreMsg msg) { public void process(CalculatedFieldStateRestoreMsg msg) {
log.info("[{}] [{}] Processing CF state restore msg.", msg.getId().entityId(), msg.getId().cfId());
states.put(msg.getId().cfId(), msg.getState()); states.put(msg.getId().cfId(), msg.getState());
} }
public void process(EntityInitCalculatedFieldMsg msg) { public void process(EntityInitCalculatedFieldMsg msg) {
log.info("[{}] Processing entity init CF msg.", msg.getCtx().getCfId());
var cfCtx = msg.getCtx(); var cfCtx = msg.getCtx();
if (msg.isForceReinit()) { if (msg.isForceReinit()) {
log.info("Force reinitialization of CF: [{}].", cfCtx.getCfId());
states.remove(cfCtx.getCfId()); states.remove(cfCtx.getCfId());
} }
var cfState = getOrInitState(cfCtx); var cfState = getOrInitState(cfCtx);
@ -96,6 +99,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} }
public void process(CalculatedFieldEntityDeleteMsg msg) { public void process(CalculatedFieldEntityDeleteMsg msg) {
log.info("[{}] Processing CF entity delete msg.", msg.getEntityId());
if (this.entityId.equals(msg.getEntityId())) { if (this.entityId.equals(msg.getEntityId())) {
MultipleTbCallback multipleTbCallback = new MultipleTbCallback(states.size(), msg.getCallback()); MultipleTbCallback multipleTbCallback = new MultipleTbCallback(states.size(), msg.getCallback());
states.forEach((cfId, state) -> cfService.deleteStateFromStorage(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback)); states.forEach((cfId, state) -> cfService.deleteStateFromStorage(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback));
@ -110,6 +114,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} }
public void process(EntityCalculatedFieldTelemetryMsg msg) { public void process(EntityCalculatedFieldTelemetryMsg msg) {
log.info("[{}] Processing CF telemetry msg.", msg.getEntityId());
var proto = msg.getProto(); var proto = msg.getProto();
var numberOfCallbacks = CALLBACKS_PER_CF * (msg.getEntityIdFields().size() + msg.getProfileIdFields().size()); var numberOfCallbacks = CALLBACKS_PER_CF * (msg.getEntityIdFields().size() + msg.getProfileIdFields().size());
MultipleTbCallback callback = new MultipleTbCallback(numberOfCallbacks, msg.getCallback()); MultipleTbCallback callback = new MultipleTbCallback(numberOfCallbacks, msg.getCallback());
@ -124,6 +129,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} }
public void process(EntityCalculatedFieldLinkedTelemetryMsg msg) { public void process(EntityCalculatedFieldLinkedTelemetryMsg msg) {
log.info("[{}] Processing CF link telemetry msg.", msg.getEntityId());
var proto = msg.getProto(); var proto = msg.getProto();
var ctx = msg.getCtx(); var ctx = msg.getCtx();
var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback()); var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback());
@ -169,6 +175,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback, private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback,
Map<String, ArgumentEntry> newArgValues, UUID tbMsgId, TbMsgType tbMsgType) { Map<String, ArgumentEntry> newArgValues, UUID tbMsgId, TbMsgType tbMsgType) {
if (newArgValues.isEmpty()) { if (newArgValues.isEmpty()) {
log.info("[{}] No new argument values to process for CF.", ctx.getCfId());
callback.onSuccess(CALLBACKS_PER_CF); callback.onSuccess(CALLBACKS_PER_CF);
} }
CalculatedFieldState state = getOrInitState(ctx); CalculatedFieldState state = getOrInitState(ctx);

28
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java

@ -91,11 +91,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
} }
public void onFieldInitMsg(CalculatedFieldInitMsg msg) { public void onFieldInitMsg(CalculatedFieldInitMsg msg) {
log.info("[{}] Processing CF init message.", msg.getCf().getId());
var cf = msg.getCf(); var cf = msg.getCf();
var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService()); var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService());
try { try {
cfCtx.init(); cfCtx.init();
} catch (Exception e) { } catch (Exception e) {
log.debug("[{}] Failed to initialize CF context.", cf.getId(), e);
if (DebugModeUtil.isDebugAllAvailable(cf)) { if (DebugModeUtil.isDebugAllAvailable(cf)) {
systemContext.persistCalculatedFieldDebugEvent(cf.getTenantId(), cf.getId(), cf.getEntityId(), null, null, null, null, e); systemContext.persistCalculatedFieldDebugEvent(cf.getTenantId(), cf.getId(), cf.getEntityId(), null, null, null, null, e);
} }
@ -108,6 +110,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
} }
public void onLinkInitMsg(CalculatedFieldLinkInitMsg msg) { public void onLinkInitMsg(CalculatedFieldLinkInitMsg msg) {
log.info("[{}] Processing CF link init message for entity [{}].", msg.getLink().getCalculatedFieldId(), msg.getLink().getEntityId());
var link = msg.getLink(); var link = msg.getLink();
// We use copy on write lists to safely pass the reference to another actor for the iteration. // We use copy on write lists to safely pass the reference to another actor for the iteration.
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
@ -121,6 +124,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
if (calculatedField != null) { if (calculatedField != null) {
msg.getState().setRequiredArguments(calculatedField.getArgNames()); msg.getState().setRequiredArguments(calculatedField.getArgNames());
log.info("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId());
getOrCreateActor(msg.getId().entityId()).tell(msg); getOrCreateActor(msg.getId().entityId()).tell(msg);
} else { } else {
cfExecService.deleteStateFromStorage(msg.getId(), msg.getCallback()); cfExecService.deleteStateFromStorage(msg.getId(), msg.getCallback());
@ -128,6 +132,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
} }
public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) { public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) {
log.info("Processing entity lifecycle event: [{}] for entity: [{}]", msg.getData().getEvent(), msg.getData().getEntityId());
var entityType = msg.getData().getEntityId().getEntityType(); var entityType = msg.getData().getEntityId().getEntityType();
var event = msg.getData().getEvent(); var event = msg.getData().getEvent();
switch (entityType) { switch (entityType) {
@ -207,6 +212,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) {
cfEntityCache.evict(tenantId, msg.getEntityId()); cfEntityCache.evict(tenantId, msg.getEntityId());
log.info("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId());
getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback));
} }
@ -225,6 +231,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
try { try {
cfCtx.init(); cfCtx.init();
} catch (Exception e) { } catch (Exception e) {
log.debug("[{}] Failed to initialize CF context.", cf.getId(), e);
if (DebugModeUtil.isDebugAllAvailable(cf)) { if (DebugModeUtil.isDebugAllAvailable(cf)) {
systemContext.persistCalculatedFieldDebugEvent(cf.getTenantId(), cf.getId(), cf.getEntityId(), null, null, null, null, e); systemContext.persistCalculatedFieldDebugEvent(cf.getTenantId(), cf.getId(), cf.getEntityId(), null, null, null, null, e);
} }
@ -251,6 +258,14 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
callback.onSuccess(); callback.onSuccess();
} else { } else {
var newCfCtx = new CalculatedFieldCtx(newCf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService()); var newCfCtx = new CalculatedFieldCtx(newCf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService());
try {
newCfCtx.init();
} catch (Exception e) {
log.debug("[{}] Failed to initialize CF context.", newCf.getId(), e);
if (DebugModeUtil.isDebugAllAvailable(newCf)) {
systemContext.persistCalculatedFieldDebugEvent(newCf.getTenantId(), newCf.getId(), newCf.getEntityId(), null, null, null, null, e);
}
}
calculatedFields.put(newCf.getId(), newCfCtx); calculatedFields.put(newCf.getId(), newCfCtx);
List<CalculatedFieldCtx> oldCfList = entityIdCalculatedFields.get(newCf.getEntityId()); List<CalculatedFieldCtx> oldCfList = entityIdCalculatedFields.get(newCf.getEntityId());
List<CalculatedFieldCtx> newCfList = new ArrayList<>(oldCfList.size()); List<CalculatedFieldCtx> newCfList = new ArrayList<>(oldCfList.size());
@ -318,12 +333,14 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
public void onTelemetryMsg(CalculatedFieldTelemetryMsg msg) { public void onTelemetryMsg(CalculatedFieldTelemetryMsg msg) {
EntityId entityId = msg.getEntityId(); EntityId entityId = msg.getEntityId();
log.info("Received telemetry msg from entity [{}]", entityId);
// 2 = 1 for CF processing + 1 for links processing // 2 = 1 for CF processing + 1 for links processing
MultipleTbCallback callback = new MultipleTbCallback(2, msg.getCallback()); MultipleTbCallback callback = new MultipleTbCallback(2, msg.getCallback());
// process all cfs related to entity, or it's profile; // process all cfs related to entity, or it's profile;
var entityIdFields = getCalculatedFieldsByEntityId(entityId); var entityIdFields = getCalculatedFieldsByEntityId(entityId);
var profileIdFields = getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId)); var profileIdFields = getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId));
if (!entityIdFields.isEmpty() || !profileIdFields.isEmpty()) { if (!entityIdFields.isEmpty() || !profileIdFields.isEmpty()) {
log.info("Pushing telemetry msg to specific actor [{}]", entityId);
getOrCreateActor(entityId).tell(new EntityCalculatedFieldTelemetryMsg(msg, entityIdFields, profileIdFields, callback)); getOrCreateActor(entityId).tell(new EntityCalculatedFieldTelemetryMsg(msg, entityIdFields, profileIdFields, callback));
} else { } else {
callback.onSuccess(); callback.onSuccess();
@ -340,6 +357,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
public void onLinkedTelemetryMsg(CalculatedFieldLinkedTelemetryMsg msg) { public void onLinkedTelemetryMsg(CalculatedFieldLinkedTelemetryMsg msg) {
EntityId sourceEntityId = msg.getEntityId(); EntityId sourceEntityId = msg.getEntityId();
log.info("Received linked telemetry msg from entity [{}]", sourceEntityId);
var proto = msg.getProto(); var proto = msg.getProto();
var linksList = proto.getLinksList(); var linksList = proto.getLinksList();
for (var linkProto : linksList) { for (var linkProto : linksList) {
@ -353,12 +371,15 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
if (!entityIds.isEmpty()) { if (!entityIds.isEmpty()) {
MultipleTbCallback callback = new MultipleTbCallback(entityIds.size(), msg.getCallback()); MultipleTbCallback callback = new MultipleTbCallback(entityIds.size(), msg.getCallback());
var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, callback); var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, callback);
entityIds.forEach(entityId -> getOrCreateActor(entityId).tell(newMsg)); entityIds.forEach(entityId -> {
log.info("Pushing linked telemetry msg to specific actor [{}]", entityId);
getOrCreateActor(entityId).tell(newMsg);
});
} else { } else {
msg.getCallback().onSuccess(); msg.getCallback().onSuccess();
} }
} else { } else {
// push the message to specific entity; log.info("Pushing linked telemetry msg to specific actor [{}]", targetEntityId);
var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, msg.getCallback()); var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, msg.getCallback());
getOrCreateActor(targetEntityId).tell(newMsg); getOrCreateActor(targetEntityId).tell(newMsg);
} }
@ -423,10 +444,12 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
} }
private void deleteCfForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) { private void deleteCfForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) {
log.info("Pushing delete CF msg to specific actor [{}]", entityId);
getOrCreateActor(entityId).tell(new CalculatedFieldEntityDeleteMsg(tenantId, cfId, callback)); getOrCreateActor(entityId).tell(new CalculatedFieldEntityDeleteMsg(tenantId, cfId, callback));
} }
private void initCfForEntity(EntityId entityId, CalculatedFieldCtx cfCtx, boolean forceStateReinit, TbCallback callback) { private void initCfForEntity(EntityId entityId, CalculatedFieldCtx cfCtx, boolean forceStateReinit, TbCallback callback) {
log.info("Pushing entity init CF msg to specific actor [{}]", entityId);
getOrCreateActor(entityId).tell(new EntityInitCalculatedFieldMsg(tenantId, cfCtx, callback, forceStateReinit)); getOrCreateActor(entityId).tell(new EntityInitCalculatedFieldMsg(tenantId, cfCtx, callback, forceStateReinit));
} }
@ -460,5 +483,4 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
oldLinks.forEach(link -> entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new ArrayList<>()).remove(link)); oldLinks.forEach(link -> entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new ArrayList<>()).remove(link));
} }
} }

Loading…
Cancel
Save