diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 71fa8fa958..e6f541090c 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -42,10 +42,12 @@ import org.thingsboard.script.api.tbel.TbelInvokeService; import org.thingsboard.server.actors.service.ActorService; import org.thingsboard.server.actors.tenant.DebugTbRateLimits; import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.data.event.CalculatedFieldDebugEvent; import org.thingsboard.server.common.data.event.ErrorEvent; import org.thingsboard.server.common.data.event.LifecycleEvent; import org.thingsboard.server.common.data.event.RuleChainDebugEvent; import org.thingsboard.server.common.data.event.RuleNodeDebugEvent; +import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; @@ -125,6 +127,7 @@ import org.thingsboard.server.service.transport.TbCoreToTransportService; import java.io.PrintWriter; import java.io.StringWriter; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledExecutorService; @@ -157,6 +160,18 @@ public class ActorSystemContext { } }; + private static final FutureCallback CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK = new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void event) { + + } + + @Override + public void onFailure(Throwable th) { + log.error("Could not save debug Event for Calculated Field", th); + } + }; + private final ConcurrentMap debugPerTenantLimits = new ConcurrentHashMap<>(); public ConcurrentMap getDebugPerTenantLimits() { @@ -723,6 +738,32 @@ public class ActorSystemContext { } } + public void persistCalculatedFieldDebugEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, Map arguments, TbMsg tbMsg, Throwable error) { + if (checkLimits(tenantId, tbMsg, error)) { + try { + CalculatedFieldDebugEvent.CalculatedFieldDebugEventBuilder event = CalculatedFieldDebugEvent.builder() + .tenantId(tenantId) + .entityId(entityId.getId()) + .serviceId(getServiceId()) + .calculatedFieldId(calculatedFieldId) + .eventEntity(tbMsg.getOriginator()) + .msgId(tbMsg.getId()) + .msgType(tbMsg.getType()) + .arguments(JacksonUtil.toString(arguments)) + .result(tbMsg.getData()); + + if (error != null) { + event.error(toString(error)); + } + + ListenableFuture future = eventService.saveAsync(event.build()); + Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor()); + } catch (IllegalArgumentException ex) { + log.warn("Failed to persist calculated field debug message", ex); + } + } + } + private boolean checkLimits(TenantId tenantId, TbMsg tbMsg, Throwable error) { if (debugPerTenantEnabled) { DebugTbRateLimits debugTbRateLimits = debugPerTenantLimits.computeIfAbsent(tenantId, id -> diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 421e4efb26..54e32446bf 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import io.netty.channel.EventLoopGroup; import lombok.extern.slf4j.Slf4j; import org.bouncycastle.util.Arrays; +import org.thingsboard.common.util.DebugModeUtil; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.rule.engine.api.MailService; @@ -64,7 +65,6 @@ import org.thingsboard.server.common.data.msg.TbNodeConnectionType; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.rule.RuleNode; -import org.thingsboard.common.util.DebugModeUtil; import org.thingsboard.server.common.data.rule.RuleNodeState; import org.thingsboard.server.common.data.script.ScriptLanguage; import org.thingsboard.server.common.msg.TbActorMsg; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java index 2507546013..1ad0377e27 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java @@ -15,10 +15,12 @@ */ package org.thingsboard.server.service.cf; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; import org.thingsboard.server.common.msg.queue.TbCallback; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityUpdateMsgProto; +import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto; import org.thingsboard.server.service.cf.telemetry.CalculatedFieldTelemetryUpdateRequest; public interface CalculatedFieldExecutionService { @@ -30,18 +32,18 @@ public interface CalculatedFieldExecutionService { */ void pushRequestToQueue(TimeseriesSaveRequest request, TimeseriesSaveResult result); + void pushRequestToQueue(AttributesSaveRequest request); + // void pushEntityUpdateMsg(TransportProtos.CalculatedFieldEntityUpdateMsgProto proto, TbCallback callback); /* ===================================================== */ - void onCalculatedFieldMsg(TransportProtos.CalculatedFieldMsgProto proto, TbCallback callback); + void onCalculatedFieldLifecycleMsg(ComponentLifecycleMsgProto proto, TbCallback callback); void onTelemetryUpdate(CalculatedFieldTelemetryUpdateRequest calculatedFieldTelemetryUpdateRequest); - void onTelemetryUpdateMsg(TransportProtos.TelemetryUpdateMsgProto proto); - - void onEntityProfileChangedMsg(TransportProtos.EntityProfileUpdateMsgProto proto, TbCallback callback); +// void onTelemetryUpdateMsg(TransportProtos.TelemetryUpdateMsgProto proto); - void onProfileEntityMsg(TransportProtos.ProfileEntityMsgProto proto, TbCallback callback); + void onEntityUpdateMsg(CalculatedFieldEntityUpdateMsgProto proto, TbCallback callback); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java index 0f9b948a9d..cbd5445cdf 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java @@ -33,17 +33,15 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.cluster.TbClusterService; -import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.EntityType; -import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.ArgumentType; -import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.OutputType; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.CalculatedFieldId; @@ -74,7 +72,14 @@ import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.cf.CalculatedFieldService; import org.thingsboard.server.dao.timeseries.TimeseriesService; +import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityUpdateMsgProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; +import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleEvent; +import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgMetadata; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtx; @@ -87,15 +92,12 @@ import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry; -import org.thingsboard.server.service.cf.telemetry.CalculatedFieldAttributeUpdateRequest; import org.thingsboard.server.service.cf.telemetry.CalculatedFieldTelemetryUpdateRequest; -import org.thingsboard.server.service.cf.telemetry.CalculatedFieldTimeSeriesUpdateRequest; import org.thingsboard.server.service.partition.AbstractPartitionBasedService; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import java.util.ArrayList; -import java.util.Collections; import java.util.EnumSet; import java.util.HashMap; import java.util.List; @@ -112,7 +114,6 @@ import java.util.function.Supplier; import java.util.stream.Collectors; import static org.thingsboard.server.common.data.DataConstants.SCOPE; -import static org.thingsboard.server.common.util.ProtoUtils.toTsKvProto; @Service @Slf4j @@ -195,7 +196,15 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas () -> toCalculatedFieldTelemetryMsgProto(request, result), request.getCallback()); } - private void checkEntityAndPushToQueue(TenantId tenantId, EntityId entityId, Predicate mainEntityFilter, Predicate linkedEntityFilter, Supplier msg, FutureCallback callback) { + @Override + public void pushRequestToQueue(AttributesSaveRequest request) { + var tenantId = request.getTenantId(); + var entityId = request.getEntityId(); + checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matches(request.getEntries(), request.getScope()), cf -> cf.linkMatches(entityId, request.getEntries(), request.getScope()), + () -> toCalculatedFieldTelemetryMsgProto(request), request.getCallback()); + } + + private void checkEntityAndPushToQueue(TenantId tenantId, EntityId entityId, Predicate mainEntityFilter, Predicate linkedEntityFilter, Supplier msg, FutureCallback callback) { boolean send = checkEntityForCalculatedFields(tenantId, entityId, mainEntityFilter, linkedEntityFilter); if (send) { clusterService.pushMsgToCalculatedFields(tenantId, entityId, msg.get(), wrap(callback)); @@ -223,31 +232,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas return send; } - private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(TimeseriesSaveRequest request, TimeseriesSaveResult result) { - ////TODO: IM to push to CF queue - return null; - } - - private void processCalculatedFieldLinks(CalculatedFieldTelemetryUpdateRequest request, Map> tpiStates) { - TenantId tenantId = request.getTenantId(); - EntityId entityId = request.getEntityId(); - - calculatedFieldCache.getCalculatedFieldLinksByEntityId(entityId) - .forEach(link -> { - CalculatedFieldId calculatedFieldId = link.getCalculatedFieldId(); - CalculatedFieldCtx ctx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId, tbelInvokeService); - EntityId targetEntityId = ctx.getEntityId(); - - if (isProfileEntity(targetEntityId)) { - calculatedFieldCache.getEntitiesByProfile(tenantId, targetEntityId).forEach(entityByProfile -> { - processCalculatedFieldLink(request, entityByProfile, ctx, tpiStates); - }); - } else { - processCalculatedFieldLink(request, targetEntityId, ctx, tpiStates); - } - }); - } - @Override protected Map>> onAddedPartitions(Set addedPartitions) { var result = new HashMap>>(); @@ -318,21 +302,21 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas } @Override - public void onCalculatedFieldMsg(TransportProtos.CalculatedFieldMsgProto proto, TbCallback callback) { + public void onCalculatedFieldLifecycleMsg(ComponentLifecycleMsgProto proto, TbCallback callback) { try { TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); - CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(proto.getCalculatedFieldIdMSB(), proto.getCalculatedFieldIdLSB())); + CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); log.info("Received CalculatedFieldMsgProto for processing: tenantId=[{}], calculatedFieldId=[{}]", tenantId, calculatedFieldId); - if (proto.getDeleted()) { + ComponentLifecycleEvent event = proto.getEvent(); + if (ComponentLifecycleEvent.DELETED.equals(event)) { log.warn("Executing onCalculatedFieldDelete, calculatedFieldId=[{}]", calculatedFieldId); calculatedFieldCache.evict(calculatedFieldId); onCalculatedFieldDelete(calculatedFieldId, callback); callback.onSuccess(); } CalculatedField cf = calculatedFieldService.findById(tenantId, calculatedFieldId); - if (proto.getUpdated()) { + if (ComponentLifecycleEvent.UPDATED.equals(event)) { log.info("Executing onCalculatedFieldUpdate, calculatedFieldId=[{}]", calculatedFieldId); - calculatedFieldCache.updateCalculatedField(tenantId, calculatedFieldId); boolean shouldReinit = onCalculatedFieldUpdate(cf, callback); if (!shouldReinit) { return; @@ -341,14 +325,14 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas if (cf != null) { calculatedFieldCache.addCalculatedField(tenantId, calculatedFieldId); EntityId entityId = cf.getEntityId(); - CalculatedFieldCtx calculatedFieldCtx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId, tbelInvokeService); + CalculatedFieldCtx calculatedFieldCtx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId); switch (entityId.getEntityType()) { case ASSET, DEVICE -> { log.info("Initializing state for entity: tenantId=[{}], entityId=[{}]", tenantId, entityId); initializeStateForEntity(calculatedFieldCtx, entityId, callback); } case ASSET_PROFILE, DEVICE_PROFILE -> { - log.info("Initializing state for all entities in profile: tenantId=[{}], profileId=[{}]", tenantId, entityId); + log.info("Initializing state for all entities in profile: tenantICalculatedFieldMsgProtod=[{}], profileId=[{}]", tenantId, entityId); Map commonArguments = calculatedFieldCtx.getArguments().entrySet().stream() .filter(entry -> entry.getValue().getRefEntityId() != null) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); @@ -380,6 +364,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas if (hasSignificantChanges(oldCalculatedField, updatedCalculatedField)) { onCalculatedFieldDelete(updatedCalculatedField.getId(), callback); } else { + calculatedFieldCache.updateCalculatedField(updatedCalculatedField.getTenantId(), updatedCalculatedField.getId()); callback.onSuccess(); shouldReinit = false; } @@ -408,13 +393,9 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas } boolean entityIdChanged = !oldCalculatedField.getEntityId().equals(newCalculatedField.getEntityId()); boolean typeChanged = !oldCalculatedField.getType().equals(newCalculatedField.getType()); - CalculatedFieldConfiguration oldConfig = oldCalculatedField.getConfiguration(); - CalculatedFieldConfiguration newConfig = newCalculatedField.getConfiguration(); - boolean argumentsChanged = !oldConfig.getArguments().equals(newConfig.getArguments()); - boolean outputTypeChanged = !oldConfig.getOutput().getType().equals(newConfig.getOutput().getType()); - boolean expressionChanged = !oldConfig.getExpression().equals(newConfig.getExpression()); + boolean argumentsChanged = !oldCalculatedField.getConfiguration().getArguments().equals(newCalculatedField.getConfiguration().getArguments()); - return entityIdChanged || typeChanged || argumentsChanged || outputTypeChanged || expressionChanged; + return entityIdChanged || typeChanged || argumentsChanged; } @Override @@ -496,30 +477,30 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas } } - @Override - public void onTelemetryUpdateMsg(TransportProtos.TelemetryUpdateMsgProto proto) { - try { - CalculatedFieldTelemetryUpdateRequest request = fromProto(proto); - - if (proto.getLinksList().isEmpty()) { - onTelemetryUpdate(request); - return; - } - - proto.getLinksList().forEach(ctxIdProto -> { - CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(ctxIdProto.getCalculatedFieldIdMSB(), ctxIdProto.getCalculatedFieldIdLSB())); - CalculatedFieldCtx ctx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId, tbelInvokeService); - - Map updatedTelemetry = request.getMappedTelemetry(ctx, request.getEntityId()); - if (!updatedTelemetry.isEmpty()) { - EntityId targetEntityId = EntityIdFactory.getByTypeAndUuid(ctxIdProto.getEntityType(), new UUID(ctxIdProto.getEntityIdMSB(), ctxIdProto.getEntityIdLSB())); - executeTelemetryUpdate(ctx, targetEntityId, request.getPreviousCalculatedFieldIds(), updatedTelemetry); - } - }); - } catch (Exception e) { - log.trace("Failed to process telemetry update msg: [{}]", proto, e); - } - } +// @Override +// public void onTelemetryUpdateMsg(TransportProtos.TelemetryUpdateMsgProto proto) { +// try { +// CalculatedFieldTelemetryUpdateRequest request = fromProto(proto); +// +// if (proto.getLinksList().isEmpty()) { +// onTelemetryUpdate(request); +// return; +// } +// +// proto.getLinksList().forEach(ctxIdProto -> { +// CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(ctxIdProto.getCalculatedFieldIdMSB(), ctxIdProto.getCalculatedFieldIdLSB())); +// CalculatedFieldCtx ctx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId, tbelInvokeService); +// +// Map updatedTelemetry = request.getMappedTelemetry(ctx, request.getEntityId()); +// if (!updatedTelemetry.isEmpty()) { +// EntityId targetEntityId = EntityIdFactory.getByTypeAndUuid(ctxIdProto.getEntityType(), new UUID(ctxIdProto.getEntityIdMSB(), ctxIdProto.getEntityIdLSB())); +// executeTelemetryUpdate(ctx, targetEntityId, request.getPreviousCalculatedFieldIds(), updatedTelemetry); +// } +// }); +// } catch (Exception e) { +// log.trace("Failed to process telemetry update msg: [{}]", proto, e); +// } +// } private void executeTelemetryUpdate(CalculatedFieldCtx cfCtx, EntityId entityId, List previousCalculatedFieldIds, Map updatedTelemetry) { log.info("Received telemetry update msg: tenantId=[{}], entityId=[{}], calculatedFieldId=[{}]", cfCtx.getTenantId(), entityId, cfCtx.getCfId()); @@ -530,50 +511,41 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas } @Override - public void onEntityProfileChangedMsg(TransportProtos.EntityProfileUpdateMsgProto proto, TbCallback callback) { + public void onEntityUpdateMsg(CalculatedFieldEntityUpdateMsgProto proto, TbCallback callback) { try { TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); - EntityId oldProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())); - EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); if (tpi.isMyPartition()) { - log.info("Received EntityProfileUpdateMsgProto for processing: tenantId=[{}], entityId=[{}]", tenantId, entityId); - calculatedFieldCache.getCalculatedFieldsByEntityId(oldProfileId).forEach(cf -> clearState(cf.getId(), entityId)); - initializeStateForEntityByProfile(entityId, newProfileId, callback); - } else { - clusterService.pushMsgToRuleEngine(tpi, UUID.randomUUID(), TransportProtos.ToRuleEngineMsg.newBuilder().setEntityProfileUpdateMsg(proto).build(), null); - } - } catch (Exception e) { - log.trace("Failed to process entity type update msg: [{}]", proto, e); - } - } - - @Override - public void onProfileEntityMsg(TransportProtos.ProfileEntityMsgProto proto, TbCallback callback) { - try { - TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); - EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); - EntityId profileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getProfileIdMSB(), proto.getProfileIdLSB())); - - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); - if (tpi.isMyPartition()) { - log.info("Received ProfileEntityMsgProto for processing: tenantId=[{}], entityId=[{}]", tenantId, entityId); + log.info("Received CalculatedFieldEntityUpdateMsgProto for processing: tenantId=[{}], entityId=[{}]", tenantId, entityId); if (proto.getDeleted()) { - log.info("Executing profile entity deleted msg, tenantId=[{}], entityId=[{}]", tenantId, entityId); + log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity deleted from profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); + EntityId oldProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())); calculatedFieldCache.getCalculatedFieldsByEntityId(entityId).forEach(cf -> clearState(cf.getId(), entityId)); - calculatedFieldCache.getCalculatedFieldsByEntityId(profileId).forEach(cf -> clearState(cf.getId(), entityId)); - } else { - log.info("Executing profile entity added msg, tenantId=[{}], entityId=[{}]", tenantId, entityId); - initializeStateForEntityByProfile(entityId, profileId, callback); + calculatedFieldCache.getCalculatedFieldsByEntityId(oldProfileId).forEach(cf -> clearState(cf.getId(), entityId)); + } + if (proto.getAdded()) { + log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity added to profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); + + EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); + initializeStateForEntityByProfile(entityId, newProfileId, callback); + } + if (proto.getUpdated()) { + log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity changed the profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); + + EntityId oldProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())); + EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); + + calculatedFieldCache.getCalculatedFieldsByEntityId(oldProfileId).forEach(cf -> clearState(cf.getId(), entityId)); + initializeStateForEntityByProfile(entityId, newProfileId, callback); } } else { - clusterService.pushMsgToRuleEngine(tpi, UUID.randomUUID(), TransportProtos.ToRuleEngineMsg.newBuilder().setProfileEntityMsg(proto).build(), null); + clusterService.pushNotificationToCalculatedFields(tenantId, entityId, ToCalculatedFieldNotificationMsg.newBuilder().setEntityUpdateMsg(proto).build(), null); } } catch (Exception e) { - log.trace("Failed to process profile entity msg: [{}]", proto, e); + log.trace("Failed to process entity update msg: [{}]", proto, e); } } @@ -586,7 +558,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas private void initializeStateForEntityByProfile(EntityId entityId, EntityId profileId, TbCallback callback) { calculatedFieldCache.getCalculatedFieldsByEntityId(profileId).stream() - .map(cf -> calculatedFieldCache.getCalculatedFieldCtx(cf.getId(), tbelInvokeService)) + .map(cf -> calculatedFieldCache.getCalculatedFieldCtx(cf.getId())) .forEach(cfCtx -> initializeStateForEntity(cfCtx, entityId, callback)); } @@ -779,88 +751,22 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas return Futures.transform(tsRollingFuture, tsRolling -> tsRolling == null ? TsRollingArgumentEntry.EMPTY : ArgumentEntry.createTsRollingArgument(tsRolling), calculatedFieldCallbackExecutor); } - private TransportProtos.TelemetryUpdateMsgProto buildTelemetryUpdateMsgProto(CalculatedFieldTelemetryUpdateRequest request) { - return buildTelemetryUpdateMsgProto(request, Collections.emptyList()); - } - - private TransportProtos.TelemetryUpdateMsgProto buildTelemetryUpdateMsgProto( - CalculatedFieldTelemetryUpdateRequest request, List links - ) { - TransportProtos.TelemetryUpdateMsgProto.Builder builder = TransportProtos.TelemetryUpdateMsgProto.newBuilder(); - - builder.setTenantIdMSB(request.getTenantId().getId().getMostSignificantBits()) - .setTenantIdLSB(request.getTenantId().getId().getLeastSignificantBits()) - .setEntityType(request.getEntityId().getEntityType().name()) - .setEntityIdMSB(request.getEntityId().getId().getMostSignificantBits()) - .setEntityIdLSB(request.getEntityId().getId().getLeastSignificantBits()); - - for (CalculatedFieldEntityCtxId link : links) { - builder.addLinks(toProto(link)); - } - - for (CalculatedFieldId calculatedFieldId : request.getPreviousCalculatedFieldIds()) { - builder.addPreviousCalculatedFields(toProto(calculatedFieldId)); - } - - if (request instanceof CalculatedFieldAttributeUpdateRequest attributeUpdateRequest) { - builder.setScope(attributeUpdateRequest.getScope().name()); - } - - for (KvEntry entry : request.getKvEntries()) { - TransportProtos.TelemetryProto.Builder telemetryBuilder = TransportProtos.TelemetryProto.newBuilder(); - if (request instanceof CalculatedFieldTimeSeriesUpdateRequest) { - telemetryBuilder.setTsKv(toTsKvProto((TsKvEntry) entry)); - } - if (request instanceof CalculatedFieldAttributeUpdateRequest attrRequest) { - telemetryBuilder.setAttrKv(ProtoUtils.toAttributeKvProto((AttributeKvEntry) entry, attrRequest.getScope())); - } - builder.addUpdatedTelemetry(telemetryBuilder.build()); - } - - return builder.build(); - } - - private CalculatedFieldTelemetryUpdateRequest fromProto(TransportProtos.TelemetryUpdateMsgProto proto) { - TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); - EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); - - List updatedTelemetry = proto.getUpdatedTelemetryList().stream() - .map(ProtoUtils::fromTelemetryProto) - .toList(); - - boolean attributesUpdated = StringUtils.isEmpty(proto.getScope()); - - return attributesUpdated - ? new CalculatedFieldAttributeUpdateRequest( - tenantId, entityId, AttributeScope.valueOf(proto.getScope()), updatedTelemetry, - proto.getPreviousCalculatedFieldsList().stream() - .map(cfIdProto -> new CalculatedFieldId( - new UUID(cfIdProto.getCalculatedFieldIdMSB(), cfIdProto.getCalculatedFieldIdLSB()))) - .toList()) - : new CalculatedFieldTimeSeriesUpdateRequest( - tenantId, entityId, updatedTelemetry, - proto.getPreviousCalculatedFieldsList().stream() - .map(cfIdProto -> new CalculatedFieldId( - new UUID(cfIdProto.getCalculatedFieldIdMSB(), cfIdProto.getCalculatedFieldIdLSB()))) - .toList()); - } - - private TransportProtos.CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { - return TransportProtos.CalculatedFieldEntityCtxIdProto.newBuilder() - .setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits()) - .setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits()) - .setEntityType(ctxId.entityId().getEntityType().name()) - .setEntityIdMSB(ctxId.entityId().getId().getMostSignificantBits()) - .setEntityIdLSB(ctxId.entityId().getId().getLeastSignificantBits()) - .build(); - } - - private TransportProtos.CalculatedFieldIdProto toProto(CalculatedFieldId cfId) { - return TransportProtos.CalculatedFieldIdProto.newBuilder() - .setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits()) - .setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits()) - .build(); - } +// private TransportProtos.CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { +// return TransportProtos.CalculatedFieldEntityCtxIdProto.newBuilder() +// .setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits()) +// .setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits()) +// .setEntityType(ctxId.entityId().getEntityType().name()) +// .setEntityIdMSB(ctxId.entityId().getId().getMostSignificantBits()) +// .setEntityIdLSB(ctxId.entityId().getId().getLeastSignificantBits()) +// .build(); +// } +// +// private TransportProtos.CalculatedFieldIdProto toProto(CalculatedFieldId cfId) { +// return TransportProtos.CalculatedFieldIdProto.newBuilder() +// .setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits()) +// .setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits()) +// .build(); +// } private KvEntry createDefaultKvEntry(Argument argument) { String key = argument.getRefEntityKey().getKey(); @@ -909,6 +815,51 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas }; } + private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(TimeseriesSaveRequest request, TimeseriesSaveResult result) { + ToCalculatedFieldMsg.Builder msg = ToCalculatedFieldMsg.newBuilder(); + + CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), request.getPreviousCalculatedFieldIds()); + for (TsKvEntry entry : request.getEntries()) { + telemetryMsg.addTsData(ProtoUtils.toTsKvProto(entry)); + } + msg.setTelemetryMsg(telemetryMsg.build()); + + return msg.build(); + } + + private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(AttributesSaveRequest request) { + ToCalculatedFieldMsg.Builder msg = ToCalculatedFieldMsg.newBuilder(); + + CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), request.getPreviousCalculatedFieldIds()); + telemetryMsg.setScope(AttributeScopeProto.valueOf(request.getScope().name())); + for (AttributeKvEntry entry : request.getEntries()) { + telemetryMsg.addAttrData(ProtoUtils.toProto(entry)); + } + msg.setTelemetryMsg(telemetryMsg.build()); + + return msg.build(); + } + + private CalculatedFieldTelemetryMsgProto.Builder buildTelemetryMsgProto(TenantId tenantId, EntityId entityId, List calculatedFieldIds) { + CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = CalculatedFieldTelemetryMsgProto.newBuilder(); + + telemetryMsg.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); + telemetryMsg.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); + + telemetryMsg.setEntityType(entityId.getEntityType().name()); + telemetryMsg.setEntityIdMSB(entityId.getId().getMostSignificantBits()); + telemetryMsg.setEntityIdLSB(entityId.getId().getLeastSignificantBits()); + + for (CalculatedFieldId cfId : calculatedFieldIds) { + CalculatedFieldIdProto.Builder calculatedFieldIdProto = CalculatedFieldIdProto.newBuilder(); + calculatedFieldIdProto.setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits()); + calculatedFieldIdProto.setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits()); + telemetryMsg.addPreviousCalculatedFields(calculatedFieldIdProto.build()); + } + + return telemetryMsg; + } + private static TbQueueCallback wrap(FutureCallback callback) { if (callback != null) { return new FutureCallbackWrapper(callback); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index 1a9f999497..e4abbee4cd 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.cf.ctx.state; import lombok.Data; import org.thingsboard.script.api.tbel.TbelInvokeService; +import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.cf.configuration.Argument; @@ -27,6 +28,7 @@ import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.util.TbPair; @@ -99,20 +101,35 @@ public class CalculatedFieldCtx { ); } + public boolean matches(List values, AttributeScope scope) { + return matchesAttributes(mainEntityArguments, values, scope); + } + + public boolean linkMatches(EntityId entityId, List values, AttributeScope scope) { + var map = linkedEntityArguments.get(entityId); + return map != null && matchesAttributes(map, values, scope); + } + public boolean matches(List values) { - return matches(mainEntityArguments, values); + return matchesTimeSeries(mainEntityArguments, values); } public boolean linkMatches(EntityId entityId, List values) { var map = linkedEntityArguments.get(entityId); - if (map == null) { - return false; - } else { - return matches(map, values); + return map != null && matchesTimeSeries(map, values); + } + + private static boolean matchesAttributes(Map argMap, List values, AttributeScope scope) { + for (AttributeKvEntry attrKv : values) { + ReferencedEntityKey attrKey = new ReferencedEntityKey(attrKv.getKey(), ArgumentType.ATTRIBUTE, scope); + if (argMap.containsKey(attrKey)) { + return true; + } } + return false; } - private static boolean matches(Map argMap, List values) { + private boolean matchesTimeSeries(Map argMap, List values) { for (TsKvEntry tsKv : values) { ReferencedEntityKey latestKey = new ReferencedEntityKey(tsKv.getKey(), ArgumentType.TS_LATEST, null); if (argMap.containsKey(latestKey)) { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java index ff4634f957..4eb3f0cd89 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.queue; +import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; import jakarta.annotation.PostConstruct; @@ -24,13 +25,17 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; +import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.common.data.id.CalculatedFieldId; +import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.QueueConfig; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.queue.TbQueueConsumer; @@ -157,8 +162,12 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer @Override protected void handleNotification(UUID id, TbProtoQueueMsg msg, TbCallback callback) { - ToCalculatedFieldNotificationMsg notification = msg.getValue(); - + ToCalculatedFieldNotificationMsg toCfNotification = msg.getValue(); + if (toCfNotification.hasComponentLifecycle()) { + forwardToCalculatedFieldService(toCfNotification.getComponentLifecycle(), callback); + } else if (toCfNotification.hasEntityUpdateMsg()) { + forwardToCalculatedFieldService(toCfNotification.getEntityUpdateMsg(), callback); + } callback.onSuccess(); } @@ -184,41 +193,29 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer // } // } // -// private void forwardToCalculatedFieldService(TransportProtos.CalculatedFieldMsgProto calculatedFieldMsg, TbCallback callback) { -// var tenantId = toTenantId(calculatedFieldMsg.getTenantIdMSB(), calculatedFieldMsg.getTenantIdLSB()); -// var calculatedFieldId = new CalculatedFieldId(new UUID(calculatedFieldMsg.getCalculatedFieldIdMSB(), calculatedFieldMsg.getCalculatedFieldIdLSB())); -// ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onCalculatedFieldMsg(calculatedFieldMsg, callback)); -// DonAsynchron.withCallback(future, -// __ -> callback.onSuccess(), -// t -> { -// log.warn("[{}] Failed to process calculated field message for calculated field [{}]", tenantId.getId(), calculatedFieldId.getId(), t); -// callback.onFailure(t); -// }); -// } -// -// private void forwardToCalculatedFieldService(TransportProtos.EntityProfileUpdateMsgProto profileUpdateMsg, TbCallback callback) { -// var tenantId = toTenantId(profileUpdateMsg.getTenantIdMSB(), profileUpdateMsg.getTenantIdLSB()); -// var entityId = EntityIdFactory.getByTypeAndUuid(profileUpdateMsg.getEntityType(), new UUID(profileUpdateMsg.getEntityIdMSB(), profileUpdateMsg.getEntityIdLSB())); -// ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onEntityProfileChangedMsg(profileUpdateMsg, callback)); -// DonAsynchron.withCallback(future, -// __ -> callback.onSuccess(), -// t -> { -// log.warn("[{}] Failed to process entity profile updated message for entity [{}]", tenantId.getId(), entityId.getId(), t); -// callback.onFailure(t); -// }); -// } -// -// private void forwardToCalculatedFieldService(TransportProtos.ProfileEntityMsgProto profileEntityMsgProto, TbCallback callback) { -// var tenantId = toTenantId(profileEntityMsgProto.getTenantIdMSB(), profileEntityMsgProto.getTenantIdLSB()); -// var entityId = EntityIdFactory.getByTypeAndUuid(profileEntityMsgProto.getEntityType(), new UUID(profileEntityMsgProto.getEntityIdMSB(), profileEntityMsgProto.getEntityIdLSB())); -// ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onProfileEntityMsg(profileEntityMsgProto, callback)); -// DonAsynchron.withCallback(future, -// __ -> callback.onSuccess(), -// t -> { -// log.warn("[{}] Failed to process profile entity message for entityId [{}]", tenantId.getId(), entityId.getId(), t); -// callback.onFailure(t); -// }); -// } + private void forwardToCalculatedFieldService(TransportProtos.ComponentLifecycleMsgProto msg, TbCallback callback) { + var tenantId = toTenantId(msg.getTenantIdMSB(), msg.getTenantIdLSB()); + var calculatedFieldId = new CalculatedFieldId(new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB())); + ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onCalculatedFieldLifecycleMsg(msg, callback)); + DonAsynchron.withCallback(future, + __ -> callback.onSuccess(), + t -> { + log.warn("[{}] Failed to process calculated field message for calculated field [{}]", tenantId.getId(), calculatedFieldId.getId(), t); + callback.onFailure(t); + }); + } + + private void forwardToCalculatedFieldService(TransportProtos.CalculatedFieldEntityUpdateMsgProto msg, TbCallback callback) { + var tenantId = toTenantId(msg.getTenantIdMSB(), msg.getTenantIdLSB()); + var entityId = EntityIdFactory.getByTypeAndUuid(msg.getEntityType(), new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB())); + ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onEntityUpdateMsg(msg, callback)); + DonAsynchron.withCallback(future, + __ -> callback.onSuccess(), + t -> { + log.warn("[{}] Failed to process entity updated message for entity [{}]", tenantId.getId(), entityId.getId(), t); + callback.onFailure(t); + }); + } private void throwNotHandled(Object msg, TbCallback callback) { log.warn("Message not handled: {}", msg); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index 725e0fdd79..b510b813d6 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java @@ -69,8 +69,7 @@ import org.thingsboard.server.common.msg.rule.engine.DeviceEdgeUpdateMsg; import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg; import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.edge.EdgeService; -import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; -import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.EdgeNotificationMsgProto; @@ -80,6 +79,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.QueueDeleteMsg; import org.thingsboard.server.gen.transport.TransportProtos.QueueUpdateMsg; import org.thingsboard.server.gen.transport.TransportProtos.ResourceDeleteMsg; import org.thingsboard.server.gen.transport.TransportProtos.ResourceUpdateMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; @@ -109,7 +110,6 @@ import java.util.UUID; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; -import static org.thingsboard.server.common.data.DataConstants.CF_QUEUE_NAME; import static org.thingsboard.server.common.util.ProtoUtils.toProto; import static org.thingsboard.server.queue.discovery.HashPartitionService.CALCULATED_FIELD_QUEUE_KEY; @@ -346,6 +346,13 @@ public class DefaultTbClusterService implements TbClusterService { toCoreMsgs.incrementAndGet(); } + @Override + public void pushNotificationToCalculatedFields(TenantId tenantId, EntityId entityId, ToCalculatedFieldNotificationMsg msg, TbQueueCallback callback) { + TopicPartitionInfo tpi = partitionService.resolve(CALCULATED_FIELD_QUEUE_KEY, entityId); + producerProvider.getCalculatedFieldsNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), msg), callback); + toCoreMsgs.incrementAndGet(); + } + @Override public void broadcastEntityStateChangeEvent(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent state) { log.trace("[{}] Processing {} state change event: {}", tenantId, entityId.getEntityType(), state); @@ -809,16 +816,23 @@ public class DefaultTbClusterService implements TbClusterService { } private void sendCalculatedFieldEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, boolean added, boolean updated, boolean deleted) { - TransportProtos.CalculatedFieldMsgProto.Builder builder = TransportProtos.CalculatedFieldMsgProto.newBuilder(); + ComponentLifecycleMsgProto.Builder builder = ComponentLifecycleMsgProto.newBuilder(); builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); - builder.setCalculatedFieldIdMSB(calculatedFieldId.getId().getMostSignificantBits()); - builder.setCalculatedFieldIdLSB(calculatedFieldId.getId().getLeastSignificantBits()); - builder.setAdded(added); - builder.setUpdated(updated); - builder.setDeleted(deleted); - TransportProtos.CalculatedFieldMsgProto msg = builder.build(); - pushMsgToCore(tenantId, calculatedFieldId, ToCoreMsg.newBuilder().setCalculatedFieldMsg(msg).build(), null); + builder.setEntityType(TransportProtos.EntityTypeProto.CALCULATED_FIELD); + builder.setEntityIdMSB(calculatedFieldId.getId().getMostSignificantBits()); + builder.setEntityIdLSB(calculatedFieldId.getId().getLeastSignificantBits()); + TransportProtos.ComponentLifecycleEvent event; + if (added) { + event = TransportProtos.ComponentLifecycleEvent.CREATED; + } else if (updated) { + event = TransportProtos.ComponentLifecycleEvent.UPDATED; + } else { + event = TransportProtos.ComponentLifecycleEvent.DELETED; + } + builder.setEvent(event); + + pushNotificationToCalculatedFields(tenantId, calculatedFieldId, ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycle(builder).build(), null); } private void handleEntityProfileUpdatedEvent(TenantId tenantId, EntityId entityId, EntityId oldProfileId, EntityId newProfileId) { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index c598540ff2..e3a1ca22d3 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -38,7 +38,6 @@ import org.thingsboard.server.common.data.alarm.AlarmInfo; import org.thingsboard.server.common.data.event.ErrorEvent; import org.thingsboard.server.common.data.event.Event; import org.thingsboard.server.common.data.event.LifecycleEvent; -import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; @@ -702,42 +701,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onCalculatedFieldMsg(calculatedFieldMsg, callback)); - DonAsynchron.withCallback(future, - __ -> callback.onSuccess(), - t -> { - log.warn("[{}] Failed to process calculated field message for calculated field [{}]", tenantId.getId(), calculatedFieldId.getId(), t); - callback.onFailure(t); - }); - } - - private void forwardToCalculatedFieldService(TransportProtos.EntityProfileUpdateMsgProto profileUpdateMsg, TbCallback callback) { - var tenantId = toTenantId(profileUpdateMsg.getTenantIdMSB(), profileUpdateMsg.getTenantIdLSB()); - var entityId = EntityIdFactory.getByTypeAndUuid(profileUpdateMsg.getEntityType(), new UUID(profileUpdateMsg.getEntityIdMSB(), profileUpdateMsg.getEntityIdLSB())); - ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onEntityProfileChangedMsg(profileUpdateMsg, callback)); - DonAsynchron.withCallback(future, - __ -> callback.onSuccess(), - t -> { - log.warn("[{}] Failed to process entity profile updated message for entity [{}]", tenantId.getId(), entityId.getId(), t); - callback.onFailure(t); - }); - } - - private void forwardToCalculatedFieldService(TransportProtos.ProfileEntityMsgProto profileEntityMsgProto, TbCallback callback) { - var tenantId = toTenantId(profileEntityMsgProto.getTenantIdMSB(), profileEntityMsgProto.getTenantIdLSB()); - var entityId = EntityIdFactory.getByTypeAndUuid(profileEntityMsgProto.getEntityType(), new UUID(profileEntityMsgProto.getEntityIdMSB(), profileEntityMsgProto.getEntityIdLSB())); - ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onProfileEntityMsg(profileEntityMsgProto, callback)); - DonAsynchron.withCallback(future, - __ -> callback.onSuccess(), - t -> { - log.warn("[{}] Failed to process profile entity message for entityId [{}]", tenantId.getId(), entityId.getId(), t); - callback.onFailure(t); - }); - } - private void forwardToNotificationSchedulerService(TransportProtos.NotificationSchedulerServiceMsg msg, TbCallback callback) { TenantId tenantId = toTenantId(msg.getTenantIdMSB(), msg.getTenantIdLSB()); NotificationRequestId notificationRequestId = new NotificationRequestId(new UUID(msg.getRequestIdMSB(), msg.getRequestIdLSB())); diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 71716a16e9..0f13dbb2f9 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -168,7 +168,9 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer log.trace("Executing saveInternal [{}]", request); ListenableFuture> saveFuture = attrService.save(request.getTenantId(), request.getEntityId(), request.getScope(), request.getEntries()); addMainCallback(saveFuture, request.getCallback()); - //TODO: IM to push to CF queue + DonAsynchron.withCallback(saveFuture, result -> { + calculatedFieldExecutionService.pushRequestToQueue(request); + }, safeCallback(request.getCallback()), tsCallBackExecutor); addWsCallback(saveFuture, success -> onAttributesUpdate(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getEntries(), request.isNotifyDevice())); addCallback(saveFuture, success -> calculatedFieldExecutionService.onTelemetryUpdate(new CalculatedFieldAttributeUpdateRequest(request)), tsCallBackExecutor); } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 94c1bef71c..fdd9923292 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1619,6 +1619,8 @@ queue: edge: "${TB_QUEUE_KAFKA_EDGE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" # Kafka properties for Edge event topic edge-event: "${TB_QUEUE_KAFKA_EDGE_EVENT_TOPIC_PROPERTIES:retention.ms:2592000000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" + # Kafka properties for Calculated Field topics + calculated-field: "${TB_QUEUE_KAFKA_CF_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" consumer-stats: # Prints lag between consumer group offset and last messages offset in Kafka topics enabled: "${TB_QUEUE_KAFKA_CONSUMER_STATS_ENABLED:true}" diff --git a/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java b/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java index 151d093c4f..b28bf73c89 100644 --- a/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java +++ b/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java @@ -79,6 +79,8 @@ public interface TbClusterService extends TbQueueClusterService { void pushMsgToCalculatedFields(TenantId tenantId, EntityId entityId, TransportProtos.ToCalculatedFieldMsg msg, TbQueueCallback callback); + void pushNotificationToCalculatedFields(TenantId tenantId, EntityId entityId, TransportProtos.ToCalculatedFieldNotificationMsg msg, TbQueueCallback callback); + void broadcastEntityStateChangeEvent(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent state); void onDeviceProfileChange(DeviceProfile deviceProfile, DeviceProfile oldDeviceProfile, TbQueueCallback callback); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java index e626c9d3d2..f4b92b3802 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.common.data.cf; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonSetter; import io.swagger.v3.oas.annotations.media.Schema; import lombok.Data; import lombok.EqualsAndHashCode; @@ -22,11 +24,13 @@ import lombok.Getter; import lombok.Setter; import org.thingsboard.server.common.data.BaseData; import org.thingsboard.server.common.data.ExportableEntity; +import org.thingsboard.server.common.data.HasDebugSettings; import org.thingsboard.server.common.data.HasName; import org.thingsboard.server.common.data.HasTenantId; import org.thingsboard.server.common.data.HasVersion; import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.SimpleCalculatedFieldConfiguration; +import org.thingsboard.server.common.data.debug.DebugSettings; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -36,7 +40,7 @@ import org.thingsboard.server.common.data.validation.NoXss; @Schema @Data @EqualsAndHashCode(callSuper = true) -public class CalculatedField extends BaseData implements HasName, HasTenantId, HasVersion, ExportableEntity { +public class CalculatedField extends BaseData implements HasName, HasTenantId, HasVersion, ExportableEntity, HasDebugSettings { private static final long serialVersionUID = 4491966747773381420L; @@ -50,6 +54,11 @@ public class CalculatedField extends BaseData implements HasN @Length(fieldName = "name") @Schema(description = "User defined name of the calculated field.") private String name; + @Deprecated + @Schema(description = "Enable/disable debug. ", example = "false", deprecated = true) + private boolean debugMode; + @Schema(description = "Debug settings object.") + private DebugSettings debugSettings; @Schema(description = "Version of calculated field configuration.", example = "0") private int configurationVersion; @Schema(implementation = SimpleCalculatedFieldConfiguration.class) @@ -109,4 +118,16 @@ public class CalculatedField extends BaseData implements HasN .toString(); } + // Getter is ignored for serialization + @JsonIgnore + public boolean isDebugMode() { + return debugMode; + } + + // Setter is annotated for deserialization + @JsonSetter + public void setDebugMode(boolean debugMode) { + this.debugMode = debugMode; + } + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEvent.java b/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEvent.java new file mode 100644 index 0000000000..e0599db358 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEvent.java @@ -0,0 +1,95 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.event; + +import com.fasterxml.jackson.databind.node.ObjectNode; +import lombok.Builder; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.Setter; +import lombok.ToString; +import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.EventInfo; +import org.thingsboard.server.common.data.id.CalculatedFieldId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; + +import java.util.UUID; + +@ToString +@EqualsAndHashCode(callSuper = true) +public class CalculatedFieldDebugEvent extends Event { + + private static final long serialVersionUID = -7091690784759639853L; + + @Builder + private CalculatedFieldDebugEvent(TenantId tenantId, UUID entityId, String serviceId, UUID id, long ts, + CalculatedFieldId calculatedFieldId, EntityId eventEntity, UUID msgId, + String msgType, String arguments, String result, String error) { + super(tenantId, entityId, serviceId, id, ts); + this.calculatedFieldId = calculatedFieldId; + this.eventEntity = eventEntity; + this.msgId = msgId; + this.msgType = msgType; + this.arguments = arguments; + this.result = result; + this.error = error; + } + + @Getter + private final CalculatedFieldId calculatedFieldId; + @Getter + private final EntityId eventEntity; + @Getter + private final UUID msgId; + @Getter + private final String msgType; + @Getter + @Setter + private String arguments; + @Getter + @Setter + private String result; + @Getter + @Setter + private String error; + + @Override + public EventType getType() { + return EventType.DEBUG_CALCULATED_FIELD; + } + + @Override + public EventInfo toInfo(EntityType entityType) { + EventInfo eventInfo = super.toInfo(entityType); + var json = (ObjectNode) eventInfo.getBody(); + json.put("calculatedFieldId", calculatedFieldId.toString()); + if (eventEntity != null) { + json.put("entityId", eventEntity.getId().toString()) + .put("entityType", eventEntity.getEntityType().name()); + } + if (msgId != null) { + json.put("msgId", msgId.toString()); + } + putNotNull(json, "msgType", msgType); + putNotNull(json, "arguments", arguments); + putNotNull(json, "result", result); + putNotNull(json, "error", error); + return eventInfo; + } + + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEventFilter.java b/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEventFilter.java new file mode 100644 index 0000000000..839583ef6c --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEventFilter.java @@ -0,0 +1,50 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.event; + +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; +import lombok.EqualsAndHashCode; +import org.thingsboard.server.common.data.StringUtils; + +@Data +@EqualsAndHashCode(callSuper = true) +@Schema +public class CalculatedFieldDebugEventFilter extends DebugEventFilter { + + @Schema(description = "String value representing the calculated field id in the event body", example = "ccbfa2fe-c8f5-45d8-bb37-6b61a6e02833") + protected String calculatedFieldId; + @Schema(description = "String value representing the entity id in the event body", example = "57b6bafe-d600-423c-9267-fe31e5218986") + protected String entityId; + @Schema(description = "String value representing the entity type", allowableValues = "DEVICE") + protected String entityType; + @Schema(description = "String value representing the message id in the rule engine", example = "dcf44612-2ce4-4e5d-b462-ebb9c5628228") + protected String msgId; + @Schema(description = "String value representing the message type", example = "POST_TELEMETRY_REQUEST") + protected String msgType; + + @Override + public EventType getEventType() { + return EventType.DEBUG_CALCULATED_FIELD; + } + + @Override + public boolean isNotEmpty() { + return super.isNotEmpty() || !StringUtils.isEmpty(calculatedFieldId) || !StringUtils.isEmpty(entityId) + || !StringUtils.isEmpty(entityType) || !StringUtils.isEmpty(msgId) || !StringUtils.isEmpty(msgType); + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/event/EventType.java b/common/data/src/main/java/org/thingsboard/server/common/data/event/EventType.java index 6f98e1537f..46d0e49370 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/event/EventType.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/event/EventType.java @@ -22,7 +22,8 @@ public enum EventType { LC_EVENT("lc_event", "LC_EVENT"), STATS("stats_event", "STATS"), DEBUG_RULE_NODE("rule_node_debug_event", "DEBUG_RULE_NODE", true), - DEBUG_RULE_CHAIN("rule_chain_debug_event", "DEBUG_RULE_CHAIN", true); + DEBUG_RULE_CHAIN("rule_chain_debug_event", "DEBUG_RULE_CHAIN", true), + DEBUG_CALCULATED_FIELD("cf_debug_event", "DEBUG_CALCULATED_FIELD", true); @Getter private final String table; diff --git a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java b/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java index 1b743316bd..3bace4d91f 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java @@ -22,7 +22,6 @@ import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.ApiUsageState; import org.thingsboard.server.common.data.ApiUsageStateValue; -import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceProfileProvisionType; @@ -59,7 +58,6 @@ import org.thingsboard.server.common.data.id.TenantProfileId; import org.thingsboard.server.common.data.kv.AttributeKey; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; -import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.DoubleDataEntry; import org.thingsboard.server.common.data.kv.JsonDataEntry; @@ -630,98 +628,6 @@ public class ProtoUtils { return new BaseAttributeKvEntry(entry, proto.getLastUpdateTs(), proto.hasVersion() ? proto.getVersion() : null); } - public static KvEntry fromProto(TransportProtos.TsKvProto proto) { - TransportProtos.KeyValueProto kvProto = proto.getKv(); - String key = kvProto.getKey(); - KvEntry entry = switch (kvProto.getType()) { - case BOOLEAN_V -> new BooleanDataEntry(key, kvProto.getBoolV()); - case LONG_V -> new LongDataEntry(key, kvProto.getLongV()); - case DOUBLE_V -> new DoubleDataEntry(key, kvProto.getDoubleV()); - case STRING_V -> new StringDataEntry(key, kvProto.getStringV()); - case JSON_V -> new JsonDataEntry(key, kvProto.getJsonV()); - default -> null; - }; - return new BasicTsKvEntry(proto.getTs(), entry, proto.hasVersion() ? proto.getVersion() : null); - } - -// public static KvEntry fromTelemetryProto(TransportProtos.TelemetryProto telemetryProto) { -// if (telemetryProto.hasAttrKv()) { -// return fromProto(telemetryProto.getAttrKv().getValue()); -// } else if (telemetryProto.hasTsKv()) { -// return fromProto(telemetryProto.getTsKv()); -// } else { -// throw new IllegalArgumentException("Unsupported TelemetryProto type: " + telemetryProto); -// } -// } - - public static TransportProtos.AttributeKey toAttributeKeyProto(String key, AttributeScope scope) { - TransportProtos.AttributeKey.Builder builder = TransportProtos.AttributeKey.newBuilder(); - builder.setAttributeKey(key); - switch (scope) { - case CLIENT_SCOPE: - builder.setScope(TransportProtos.AttributeScopeProto.CLIENT_SCOPE); - break; - case SERVER_SCOPE: - builder.setScope(TransportProtos.AttributeScopeProto.SERVER_SCOPE); - break; - case SHARED_SCOPE: - builder.setScope(TransportProtos.AttributeScopeProto.SHARED_SCOPE); - break; - default: - throw new IllegalArgumentException("Unsupported attribute scope: " + scope); - } - return builder.build(); - } - -// public static TransportProtos.AttributeKvProto toAttributeKvProto(AttributeKvEntry attributeKvEntry, AttributeScope scope) { -// return TransportProtos.AttributeKvProto.newBuilder() -// .setKey(ProtoUtils.toAttributeKeyProto(attributeKvEntry.getKey(), scope)) -// .setValue(ProtoUtils.toAttributeValueProto(attributeKvEntry)) -// .build(); -// } - - public static TransportProtos.AttributeValueProto toAttributeValueProto(AttributeKvEntry attributeKvEntry) { - TransportProtos.AttributeValueProto.Builder builder = TransportProtos.AttributeValueProto.newBuilder(); - builder.setLastUpdateTs(attributeKvEntry.getLastUpdateTs()); - switch (attributeKvEntry.getDataType()) { - case BOOLEAN: - builder.setType(TransportProtos.KeyValueType.BOOLEAN_V) - .setHasV(true) - .setBoolV(attributeKvEntry.getBooleanValue().orElse(false)); - break; - case LONG: - builder.setType(TransportProtos.KeyValueType.LONG_V) - .setHasV(true) - .setLongV(attributeKvEntry.getLongValue().orElse(0L)); - break; - case DOUBLE: - builder.setType(TransportProtos.KeyValueType.DOUBLE_V) - .setHasV(true) - .setDoubleV(attributeKvEntry.getDoubleValue().orElse(0.0)); - break; - case STRING: - builder.setType(TransportProtos.KeyValueType.STRING_V) - .setHasV(true) - .setStringV(attributeKvEntry.getStrValue().orElse("")); - break; - case JSON: - builder.setType(TransportProtos.KeyValueType.JSON_V) - .setHasV(true) - .setJsonV(attributeKvEntry.getJsonValue().orElse("{}")); - break; - default: - builder.setHasV(false); - throw new IllegalArgumentException("Unsupported AttributeKvEntry data type: " + attributeKvEntry.getDataType()); - } - if (attributeKvEntry.getKey() != null) { - builder.setKey(attributeKvEntry.getKey()); - } - if (attributeKvEntry.getVersion() != null) { - builder.setVersion(attributeKvEntry.getVersion()); - } - return builder.build(); - } - public static TransportProtos.TsKvProto toTsKvProto(TsKvEntry tsKvEntry) { return TransportProtos.TsKvProto.newBuilder() .setTs(tsKvEntry.getTs()) diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java index ee529e8a68..cdd0add38b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java @@ -52,6 +52,8 @@ public class TbKafkaTopicConfigs { private String housekeeperProperties; @Value("${queue.kafka.topic-properties.housekeeper-reprocessing:}") private String housekeeperReprocessingProperties; + @Value("${queue.kafka.topic-properties.calculated-field:}") + private String calculatedFieldProperties; @Getter private Map coreConfigs; @@ -79,6 +81,8 @@ public class TbKafkaTopicConfigs { private Map edgeConfigs; @Getter private Map edgeEventConfigs; + @Getter + private Map calculatedFieldConfigs; @PostConstruct private void init() { @@ -97,6 +101,7 @@ public class TbKafkaTopicConfigs { housekeeperReprocessingConfigs = PropertyUtils.getProps(housekeeperReprocessingProperties); edgeConfigs = PropertyUtils.getProps(edgeProperties); edgeEventConfigs = PropertyUtils.getProps(edgeEventProperties); + calculatedFieldConfigs = PropertyUtils.getProps(calculatedFieldProperties); } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index dd5d61e834..01e174023c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -25,6 +25,9 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; @@ -54,6 +57,7 @@ import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; +import org.thingsboard.server.queue.settings.TbQueueCalculatedFieldSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueEdgeSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; @@ -79,6 +83,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueVersionControlSettings vcSettings; private final TbQueueEdgeSettings edgeSettings; + private final TbQueueCalculatedFieldSettings calculatedFieldSettings; private final TbKafkaConsumerStatsService consumerStatsService; private final TbQueueAdmin coreAdmin; @@ -94,6 +99,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi private final TbQueueAdmin housekeeperReprocessingAdmin; private final TbQueueAdmin edgeAdmin; private final TbQueueAdmin edgeEventAdmin; + private final TbQueueAdmin cfAdmin; private final AtomicLong consumerCount = new AtomicLong(); @@ -106,6 +112,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueVersionControlSettings vcSettings, TbQueueEdgeSettings edgeSettings, + TbQueueCalculatedFieldSettings calculatedFieldSettings, TbKafkaConsumerStatsService consumerStatsService, TbKafkaTopicConfigs kafkaTopicConfigs) { this.topicService = topicService; @@ -119,6 +126,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi this.vcSettings = vcSettings; this.consumerStatsService = consumerStatsService; this.edgeSettings = edgeSettings; + this.calculatedFieldSettings = calculatedFieldSettings; this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); @@ -133,6 +141,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi this.housekeeperReprocessingAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getHousekeeperReprocessingConfigs()); this.edgeAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeConfigs()); this.edgeEventAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs()); + this.cfAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCalculatedFieldConfigs()); } @Override @@ -490,6 +499,75 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi return requestBuilder.build(); } + @Override + public TbQueueConsumer> createToCalculatedFieldMsgConsumer() { + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); + consumerBuilder.settings(kafkaSettings); + consumerBuilder.topic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); + consumerBuilder.clientId("monolith-calculated-field-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); + consumerBuilder.groupId(topicService.buildTopicName("monolith-calculated-field-consumer")); + consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCalculatedFieldMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(cfAdmin); + consumerBuilder.statsService(consumerStatsService); + return consumerBuilder.build(); + } + + @Override + public TbQueueProducer> createToCalculatedFieldMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("monolith-calculated-field-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); + requestBuilder.admin(cfAdmin); + return requestBuilder.build(); + } + + @Override + public TbQueueConsumer> createToCalculatedFieldNotificationsMsgConsumer() { + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); + consumerBuilder.settings(kafkaSettings); + consumerBuilder.topic(topicService.getCalculatedFieldNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); + consumerBuilder.clientId("monolith-calculated-field-notifications-consumer-" + serviceInfoProvider.getServiceId()); + consumerBuilder.groupId(topicService.buildTopicName("monolith-calculated-field-notifications-consumer-" + serviceInfoProvider.getServiceId())); + consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCalculatedFieldNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(notificationAdmin); + consumerBuilder.statsService(consumerStatsService); + return consumerBuilder.build(); + } + + @Override + public TbQueueProducer> createToCalculatedFieldNotificationMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("monolith-calculated-field-notifications-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); + requestBuilder.admin(notificationAdmin); + return requestBuilder.build(); + } + + @Override + public TbQueueConsumer> createCalculatedFieldStateConsumer() { + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); + consumerBuilder.settings(kafkaSettings); + consumerBuilder.topic(topicService.buildTopicName(calculatedFieldSettings.getStateTopic())); + consumerBuilder.clientId("monolith-calculated-field-state-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); + consumerBuilder.groupId(topicService.buildTopicName("monolith-calculated-field-state-consumer")); + consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), CalculatedFieldStateProto.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(cfAdmin); + consumerBuilder.statsService(consumerStatsService); + return consumerBuilder.build(); + } + + @Override + public TbQueueProducer> createCalculatedFieldStateProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("monolith-calculated-field-state-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(topicService.buildTopicName(calculatedFieldSettings.getStateTopic())); + requestBuilder.admin(cfAdmin); + return requestBuilder.build(); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { @@ -522,5 +600,8 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi if (edgeAdmin != null) { edgeAdmin.destroy(); } + if (cfAdmin != null) { + cfAdmin.destroy(); + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index cc0e044917..329f1d7544 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -24,6 +24,8 @@ import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; @@ -53,6 +55,7 @@ import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; +import org.thingsboard.server.queue.settings.TbQueueCalculatedFieldSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueEdgeSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; @@ -79,6 +82,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { private final TbKafkaConsumerStatsService consumerStatsService; private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueEdgeSettings edgeSettings; + private final TbQueueCalculatedFieldSettings calculatedFieldSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -93,6 +97,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueAdmin housekeeperReprocessingAdmin; private final TbQueueAdmin edgeAdmin; private final TbQueueAdmin edgeEventAdmin; + private final TbQueueAdmin cfAdmin; private final AtomicLong consumerCount = new AtomicLong(); @@ -107,6 +112,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { TbQueueEdgeSettings edgeSettings, TbKafkaConsumerStatsService consumerStatsService, TbQueueTransportNotificationSettings transportNotificationSettings, + TbQueueCalculatedFieldSettings calculatedFieldSettings, TbKafkaTopicConfigs kafkaTopicConfigs) { this.topicService = topicService; this.kafkaSettings = kafkaSettings; @@ -119,6 +125,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { this.consumerStatsService = consumerStatsService; this.transportNotificationSettings = transportNotificationSettings; this.edgeSettings = edgeSettings; + this.calculatedFieldSettings = calculatedFieldSettings; this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); @@ -133,6 +140,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { this.housekeeperReprocessingAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getHousekeeperReprocessingConfigs()); this.edgeAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeConfigs()); this.edgeEventAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs()); + this.cfAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCalculatedFieldConfigs()); } @Override @@ -439,6 +447,26 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { return requestBuilder.build(); } + @Override + public TbQueueProducer> createToCalculatedFieldMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("tb-core-to-calculated-field-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); + requestBuilder.admin(cfAdmin); + return requestBuilder.build(); + } + + @Override + public TbQueueProducer> createToCalculatedFieldNotificationMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("tb-core-calculated-field-notifications-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); + requestBuilder.admin(notificationAdmin); + return requestBuilder.build(); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { @@ -468,5 +496,8 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { if (vcAdmin != null) { vcAdmin.destroy(); } + if (cfAdmin != null) { + cfAdmin.destroy(); + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java index 87a1a69c2e..46b35f9acd 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java @@ -23,6 +23,9 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; @@ -49,6 +52,7 @@ import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; +import org.thingsboard.server.queue.settings.TbQueueCalculatedFieldSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueEdgeSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; @@ -71,6 +75,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { private final TbKafkaConsumerStatsService consumerStatsService; private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueEdgeSettings edgeSettings; + private final TbQueueCalculatedFieldSettings calculatedFieldSettings; private final TbQueueAdmin coreAdmin; private final TbKafkaAdmin ruleEngineAdmin; @@ -81,6 +86,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { private final TbQueueAdmin housekeeperAdmin; private final TbQueueAdmin edgeAdmin; private final TbQueueAdmin edgeEventAdmin; + private final TbQueueAdmin cfAdmin; private final AtomicLong consumerCount = new AtomicLong(); public KafkaTbRuleEngineQueueFactory(TopicService topicService, TbKafkaSettings kafkaSettings, @@ -90,7 +96,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbKafkaConsumerStatsService consumerStatsService, TbQueueTransportNotificationSettings transportNotificationSettings, - TbQueueEdgeSettings edgeSettings, + TbQueueEdgeSettings edgeSettings, TbQueueCalculatedFieldSettings calculatedFieldSettings, TbKafkaTopicConfigs kafkaTopicConfigs) { this.topicService = topicService; this.kafkaSettings = kafkaSettings; @@ -101,6 +107,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { this.consumerStatsService = consumerStatsService; this.transportNotificationSettings = transportNotificationSettings; this.edgeSettings = edgeSettings; + this.calculatedFieldSettings = calculatedFieldSettings; this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); @@ -111,6 +118,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { this.housekeeperAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getHousekeeperConfigs()); this.edgeAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeConfigs()); this.edgeEventAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs()); + this.cfAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCalculatedFieldConfigs()); } @Override @@ -293,6 +301,65 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { .build(); } + @Override + public TbQueueConsumer> createToCalculatedFieldMsgConsumer() { + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); + consumerBuilder.settings(kafkaSettings); + consumerBuilder.topic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); + consumerBuilder.clientId("tb-rule-engine-calculated-field-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); + consumerBuilder.groupId(topicService.buildTopicName("tb-rule-engine-calculated-field-consumer")); + consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCalculatedFieldMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(cfAdmin); + consumerBuilder.statsService(consumerStatsService); + return consumerBuilder.build(); + } + + @Override + public TbQueueProducer> createToCalculatedFieldMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("tb-rule-engine-to-calculated-field-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); + requestBuilder.admin(cfAdmin); + return requestBuilder.build(); + } + + @Override + public TbQueueConsumer> createToCalculatedFieldNotificationsMsgConsumer() { + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); + consumerBuilder.settings(kafkaSettings); + consumerBuilder.topic(topicService.getCalculatedFieldNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); + consumerBuilder.clientId("tb-calculated-field-notifications-consumer-" + serviceInfoProvider.getServiceId()); + consumerBuilder.groupId(topicService.buildTopicName("tb-calculated-field-notifications-node-") + serviceInfoProvider.getServiceId()); + consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCalculatedFieldNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(notificationAdmin); + consumerBuilder.statsService(consumerStatsService); + return consumerBuilder.build(); + } + + @Override + public TbQueueConsumer> createCalculatedFieldStateConsumer() { + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); + consumerBuilder.settings(kafkaSettings); + consumerBuilder.topic(topicService.buildTopicName(calculatedFieldSettings.getStateTopic())); + consumerBuilder.clientId("tb-rule-engine-calculated-field-state-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); + consumerBuilder.groupId(topicService.buildTopicName("tb-rule-engine-calculated-field-state-consumer")); + consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), CalculatedFieldStateProto.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(cfAdmin); + consumerBuilder.statsService(consumerStatsService); + return consumerBuilder.build(); + } + + @Override + public TbQueueProducer> createCalculatedFieldStateProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("tb-rule-engine-to-calculated-field-state-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); + requestBuilder.admin(cfAdmin); + return requestBuilder.build(); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { @@ -313,5 +380,8 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { if (fwUpdatesAdmin != null) { fwUpdatesAdmin.destroy(); } + if (cfAdmin != null) { + cfAdmin.destroy(); + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java index 69b3dff1f0..1ece8249f2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java @@ -18,7 +18,7 @@ package org.thingsboard.server.queue.provider; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.gen.js.JsInvokeProtos; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; @@ -161,7 +161,7 @@ public interface TbCoreQueueFactory extends TbUsageStatsClientQueueFactory, Hous return null; } - TbQueueProducer> createToCalculatedFieldMsgProducer(); + TbQueueProducer> createToCalculatedFieldMsgProducer(); TbQueueProducer> createToCalculatedFieldNotificationMsgProducer(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java b/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java index 9c81d91f64..26ed4134cc 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java @@ -64,6 +64,7 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements log.trace("Executing save calculated field, [{}]", calculatedField); CalculatedField savedCalculatedField = calculatedFieldDao.save(tenantId, calculatedField); createOrUpdateCalculatedFieldLink(tenantId, savedCalculatedField); + updateDebugSettings(tenantId, calculatedField, System.currentTimeMillis()); eventPublisher.publishEvent(SaveEntityEvent.builder().tenantId(savedCalculatedField.getTenantId()).entityId(savedCalculatedField.getId()) .entity(savedCalculatedField).oldEntity(oldCalculatedField).created(calculatedField.getId() == null).build()); return savedCalculatedField; diff --git a/dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java b/dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java index 5314dcd405..dc57b07f72 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java @@ -23,6 +23,7 @@ import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EventInfo; import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.event.CalculatedFieldDebugEvent; import org.thingsboard.server.common.data.event.ErrorEvent; import org.thingsboard.server.common.data.event.Event; import org.thingsboard.server.common.data.event.EventFilter; @@ -88,6 +89,11 @@ public class BaseEventService implements EventService { ErrorEvent eEvent = (ErrorEvent) event; truncateField(eEvent, ErrorEvent::getError, ErrorEvent::setError); break; + case DEBUG_CALCULATED_FIELD: + CalculatedFieldDebugEvent cfEvent = (CalculatedFieldDebugEvent) event; + truncateField(cfEvent, CalculatedFieldDebugEvent::getArguments, CalculatedFieldDebugEvent::setArguments); + truncateField(cfEvent, CalculatedFieldDebugEvent::getResult, CalculatedFieldDebugEvent::setResult); + truncateField(cfEvent, CalculatedFieldDebugEvent::getError, CalculatedFieldDebugEvent::setError); } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java index e07e36bb96..9b1dc69c15 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java @@ -375,6 +375,7 @@ public class ModelConstants { public static final String STATS_EVENT_TABLE_NAME = "stats_event"; public static final String RULE_NODE_DEBUG_EVENT_TABLE_NAME = "rule_node_debug_event"; public static final String RULE_CHAIN_DEBUG_EVENT_TABLE_NAME = "rule_chain_debug_event"; + public static final String CALCULATED_FIELD_DEBUG_EVENT_TABLE_NAME = "cf_debug_event"; public static final String EVENT_TENANT_ID_PROPERTY = TENANT_ID_PROPERTY; public static final String EVENT_SERVICE_ID_PROPERTY = "service_id"; @@ -399,6 +400,10 @@ public class ModelConstants { public static final String EVENT_METADATA_COLUMN_NAME = "e_metadata"; public static final String EVENT_MESSAGE_COLUMN_NAME = "e_message"; + public static final String EVENT_CALCULATED_FIELD_ID_COLUMN_NAME = "cf_id"; + public static final String EVENT_CALCULATED_FIELD_ARGUMENTS_COLUMN_NAME = "e_args"; + public static final String EVENT_CALCULATED_FIELD_RESULT_COLUMN_NAME = "e_result"; + public static final String DEBUG_MODE = "debug_mode"; public static final String DEBUG_SETTINGS = "debug_settings"; public static final String SINGLETON_MODE = "singleton_mode"; diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldDebugEventEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldDebugEventEntity.java new file mode 100644 index 0000000000..57849ae3ea --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldDebugEventEntity.java @@ -0,0 +1,104 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.model.sql; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Table; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.NoArgsConstructor; +import org.thingsboard.server.common.data.event.CalculatedFieldDebugEvent; +import org.thingsboard.server.common.data.id.CalculatedFieldId; +import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.dao.model.BaseEntity; + +import java.util.UUID; + +import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_DEBUG_EVENT_TABLE_NAME; +import static org.thingsboard.server.dao.model.ModelConstants.EVENT_CALCULATED_FIELD_ARGUMENTS_COLUMN_NAME; +import static org.thingsboard.server.dao.model.ModelConstants.EVENT_CALCULATED_FIELD_ID_COLUMN_NAME; +import static org.thingsboard.server.dao.model.ModelConstants.EVENT_CALCULATED_FIELD_RESULT_COLUMN_NAME; +import static org.thingsboard.server.dao.model.ModelConstants.EVENT_ENTITY_ID_COLUMN_NAME; +import static org.thingsboard.server.dao.model.ModelConstants.EVENT_ENTITY_TYPE_COLUMN_NAME; +import static org.thingsboard.server.dao.model.ModelConstants.EVENT_ERROR_COLUMN_NAME; +import static org.thingsboard.server.dao.model.ModelConstants.EVENT_MSG_ID_COLUMN_NAME; +import static org.thingsboard.server.dao.model.ModelConstants.EVENT_MSG_TYPE_COLUMN_NAME; + +@Data +@EqualsAndHashCode(callSuper = true) +@Entity +@Table(name = CALCULATED_FIELD_DEBUG_EVENT_TABLE_NAME) +@NoArgsConstructor +public class CalculatedFieldDebugEventEntity extends EventEntity implements BaseEntity { + + @Column(name = EVENT_CALCULATED_FIELD_ID_COLUMN_NAME) + private UUID calculatedFieldId; + @Column(name = EVENT_ENTITY_ID_COLUMN_NAME) + private UUID eventEntityId; + @Column(name = EVENT_ENTITY_TYPE_COLUMN_NAME) + private String eventEntityType; + @Column(name = EVENT_MSG_ID_COLUMN_NAME) + private UUID msgId; + @Column(name = EVENT_MSG_TYPE_COLUMN_NAME) + private String msgType; + @Column(name = EVENT_CALCULATED_FIELD_ARGUMENTS_COLUMN_NAME) + private String arguments; + @Column(name = EVENT_CALCULATED_FIELD_RESULT_COLUMN_NAME) + private String result; + @Column(name = EVENT_ERROR_COLUMN_NAME) + private String error; + + public CalculatedFieldDebugEventEntity(CalculatedFieldDebugEvent event) { + super(event); + if (event.getCalculatedFieldId() != null) { + this.calculatedFieldId = event.getCalculatedFieldId().getId(); + } + if (event.getEventEntity() != null) { + this.eventEntityId = event.getEventEntity().getId(); + this.eventEntityType = event.getEventEntity().getEntityType().name(); + } + this.msgId = event.getMsgId(); + this.msgType = event.getMsgType(); + this.arguments = event.getArguments(); + this.result = event.getResult(); + this.error = event.getError(); + } + + @Override + public CalculatedFieldDebugEvent toData() { + var builder = CalculatedFieldDebugEvent.builder() + .id(id) + .tenantId(TenantId.fromUUID(tenantId)) + .ts(ts) + .serviceId(serviceId) + .entityId(entityId) + .msgId(msgId) + .msgType(msgType) + .arguments(arguments) + .result(result) + .error(error); + if (calculatedFieldId != null) { + builder.calculatedFieldId(new CalculatedFieldId(calculatedFieldId)); + } + if (eventEntityId != null) { + builder.eventEntity(EntityIdFactory.getByTypeAndUuid(eventEntityType, eventEntityId)); + } + return builder.build(); + } + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldEntity.java index a0157cde66..64c8e8d5b8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldEntity.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldEntity.java @@ -26,6 +26,7 @@ import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration; +import org.thingsboard.server.common.data.debug.DebugSettings; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.TenantId; @@ -45,6 +46,7 @@ import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_T import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_TENANT_ID_COLUMN; import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_TYPE; import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_VERSION; +import static org.thingsboard.server.dao.model.ModelConstants.DEBUG_SETTINGS; @Data @EqualsAndHashCode(callSuper = true) @@ -77,6 +79,9 @@ public class CalculatedFieldEntity extends BaseSqlEntity implem @Column(name = CALCULATED_FIELD_VERSION) private Long version; + @Column(name = DEBUG_SETTINGS) + private String debugSettings; + @Column(name = CALCULATED_FIELD_EXTERNAL_ID) private UUID externalId; @@ -95,6 +100,7 @@ public class CalculatedFieldEntity extends BaseSqlEntity implem this.configurationVersion = calculatedField.getConfigurationVersion(); this.configuration = JacksonUtil.valueToTree(calculatedField.getConfiguration()); this.version = calculatedField.getVersion(); + this.debugSettings = JacksonUtil.toString(calculatedField.getDebugSettings()); if (calculatedField.getExternalId() != null) { this.externalId = calculatedField.getExternalId().getId(); } @@ -111,6 +117,7 @@ public class CalculatedFieldEntity extends BaseSqlEntity implem calculatedField.setConfigurationVersion(configurationVersion); calculatedField.setConfiguration(JacksonUtil.treeToValue(configuration, CalculatedFieldConfiguration.class)); calculatedField.setVersion(version); + calculatedField.setDebugSettings(JacksonUtil.fromString(debugSettings, DebugSettings.class)); if (externalId != null) { calculatedField.setExternalId(new CalculatedFieldId(externalId)); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/CalculatedFieldDebugEventRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/CalculatedFieldDebugEventRepository.java new file mode 100644 index 0000000000..c0bd21ca74 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/CalculatedFieldDebugEventRepository.java @@ -0,0 +1,135 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.sql.event; + +import org.springframework.data.domain.Page; +import org.springframework.data.domain.Pageable; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; +import org.springframework.transaction.annotation.Transactional; +import org.thingsboard.server.common.data.event.CalculatedFieldDebugEvent; +import org.thingsboard.server.dao.model.sql.CalculatedFieldDebugEventEntity; + +import java.util.List; +import java.util.UUID; + +public interface CalculatedFieldDebugEventRepository extends EventRepository, JpaRepository { + + @Override + @Query(nativeQuery = true, value = "SELECT * FROM cf_debug_event e WHERE e.tenant_id = :tenantId AND e.entity_id = :entityId ORDER BY e.ts DESC LIMIT :limit") + List findLatestEvents(@Param("tenantId") UUID tenantId, @Param("entityId") UUID entityId, @Param("limit") int limit); + + @Override + @Query("SELECT e FROM RuleNodeDebugEventEntity e WHERE " + + "e.tenantId = :tenantId " + + "AND e.entityId = :entityId " + + "AND (:startTime IS NULL OR e.ts >= :startTime) " + + "AND (:endTime IS NULL OR e.ts <= :endTime)" + ) + Page findEvents(@Param("tenantId") UUID tenantId, + @Param("entityId") UUID entityId, + @Param("startTime") Long startTime, + @Param("endTime") Long endTime, + Pageable pageable); + + @Query(nativeQuery = true, + value = "SELECT * FROM cf_debug_event e WHERE " + + "e.tenant_id = :tenantId " + + "AND e.entity_id = :entityId " + + "AND (:startTime IS NULL OR e.ts >= :startTime) " + + "AND (:endTime IS NULL OR e.ts <= :endTime) " + + "AND (:serviceId IS NULL OR e.service_id ILIKE concat('%', :serviceId, '%')) " + + "AND (:calculatedFieldId IS NULL OR e.cf_id = uuid(:calculatedFieldId)) " + + "AND (:eventEntityId IS NULL OR e.e_entity_id = uuid(:eventEntityId)) " + + "AND (:eventEntityType IS NULL OR e.e_entity_type ILIKE concat('%', :eventEntityType, '%')) " + + "AND (:msgId IS NULL OR e.e_msg_id = uuid(:msgId)) " + + "AND (:msgType IS NULL OR e.e_msg_type ILIKE concat('%', :msgType, '%')) " + + "AND ((:isError = FALSE) OR e.e_error IS NOT NULL) " + + "AND (:error IS NULL OR e.e_error ILIKE concat('%', :error, '%'))" + , + countQuery = "SELECT count(*) FROM rule_node_debug_event e WHERE " + + "e.tenant_id = :tenantId " + + "AND e.entity_id = :entityId " + + "AND (:startTime IS NULL OR e.ts >= :startTime) " + + "AND (:endTime IS NULL OR e.ts <= :endTime) " + + "AND (:serviceId IS NULL OR e.service_id ILIKE concat('%', :serviceId, '%')) " + + "AND (:calculatedFieldId IS NULL OR e.cf_id = uuid(:calculatedFieldId)) " + + "AND (:eventEntityId IS NULL OR e.e_entity_id = uuid(:eventEntityId)) " + + "AND (:eventEntityType IS NULL OR e.e_entity_type ILIKE concat('%', :eventEntityType, '%')) " + + "AND (:msgId IS NULL OR e.e_msg_id = uuid(:msgId)) " + + "AND (:msgType IS NULL OR e.e_msg_type ILIKE concat('%', :msgType, '%')) " + + "AND ((:isError = FALSE) OR e.e_error IS NOT NULL) " + + "AND (:error IS NULL OR e.e_error ILIKE concat('%', :error, '%'))" + ) + Page findEvents(@Param("tenantId") UUID tenantId, + @Param("entityId") UUID entityId, + @Param("startTime") Long startTime, + @Param("endTime") Long endTime, + @Param("serviceId") String server, + @Param("calculatedFieldId") UUID calculatedFieldId, + @Param("eventEntityId") String eventEntityId, + @Param("eventEntityType") String eventEntityType, + @Param("msgId") String eventMsgId, + @Param("msgType") String eventMsgType, + @Param("isError") boolean isError, + @Param("error") String error, + Pageable pageable); + + @Transactional + @Modifying + @Query("DELETE FROM CalculatedFieldDebugEventEntity e WHERE " + + "e.tenantId = :tenantId " + + "AND e.entityId = :entityId " + + "AND (:startTime IS NULL OR e.ts >= :startTime) " + + "AND (:endTime IS NULL OR e.ts <= :endTime)" + ) + void removeEvents(@Param("tenantId") UUID tenantId, + @Param("entityId") UUID entityId, + @Param("startTime") Long startTime, + @Param("endTime") Long endTime); + + @Transactional + @Modifying + @Query(nativeQuery = true, + value = "DELETE FROM cf_debug_event e WHERE " + + "e.tenant_id = :tenantId " + + "AND e.entity_id = :entityId " + + "AND (:startTime IS NULL OR e.ts >= :startTime) " + + "AND (:endTime IS NULL OR e.ts <= :endTime) " + + "AND (:serviceId IS NULL OR e.service_id ILIKE concat('%', :serviceId, '%')) " + + "AND (:calculatedFieldId IS NULL OR e.cf_id = uuid(:calculatedFieldId)) " + + "AND (:eventEntityId IS NULL OR e.e_entity_id = uuid(:eventEntityId)) " + + "AND (:eventEntityType IS NULL OR e.e_entity_type ILIKE concat('%', :eventEntityType, '%')) " + + "AND (:msgId IS NULL OR e.e_msg_id = uuid(:msgId)) " + + "AND (:msgType IS NULL OR e.e_msg_type ILIKE concat('%', :msgType, '%')) " + + "AND ((:isError = FALSE) OR e.e_error IS NOT NULL) " + + "AND (:error IS NULL OR e.e_error ILIKE concat('%', :error, '%'))") + void removeEvents(@Param("tenantId") UUID tenantId, + @Param("entityId") UUID entityId, + @Param("startTime") Long startTime, + @Param("endTime") Long endTime, + @Param("serviceId") String server, + @Param("calculatedFieldId") UUID calculatedFieldId, + @Param("eventEntityId") String eventEntityId, + @Param("eventEntityType") String eventEntityType, + @Param("msgId") String eventMsgId, + @Param("msgType") String eventMsgType, + @Param("isError") boolean isError, + @Param("error") String error); + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/DedicatedJpaEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/DedicatedJpaEventDao.java index 9b7af5e7f7..7bf9b426e1 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/DedicatedJpaEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/DedicatedJpaEventDao.java @@ -36,10 +36,11 @@ public class DedicatedJpaEventDao extends JpaBaseEventDao { RuleNodeDebugEventRepository ruleNodeDebugEventRepository, RuleChainDebugEventRepository ruleChainDebugEventRepository, ScheduledLogExecutorComponent logExecutor, - StatsFactory statsFactory) { + StatsFactory statsFactory, + CalculatedFieldDebugEventRepository cfDebugEventRepository) { super(partitionConfiguration, partitioningRepository, lcEventRepository, statsEventRepository, errorEventRepository, eventInsertRepository, ruleNodeDebugEventRepository, - ruleChainDebugEventRepository, logExecutor, statsFactory); + ruleChainDebugEventRepository, logExecutor, statsFactory, cfDebugEventRepository); } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java index 962be57892..bdb675cbb7 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java @@ -25,6 +25,7 @@ import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionCallbackWithoutResult; import org.springframework.transaction.support.TransactionTemplate; +import org.thingsboard.server.common.data.event.CalculatedFieldDebugEvent; import org.thingsboard.server.common.data.event.ErrorEvent; import org.thingsboard.server.common.data.event.Event; import org.thingsboard.server.common.data.event.EventType; @@ -81,6 +82,9 @@ public class EventInsertRepository { insertStmtMap.put(EventType.DEBUG_RULE_CHAIN, "INSERT INTO " + EventType.DEBUG_RULE_CHAIN.getTable() + " (id, tenant_id, ts, entity_id, service_id, e_message, e_error) " + "VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING;"); + insertStmtMap.put(EventType.DEBUG_CALCULATED_FIELD, "INSERT INTO " + EventType.DEBUG_CALCULATED_FIELD.getTable() + + " (id, tenant_id, ts, entity_id, service_id, cf_id, e_entity_id, e_entity_type, e_msg_id, e_msg_type, e_args, e_result, e_error) " + + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING;"); } public void save(List entities) { @@ -107,6 +111,8 @@ public class EventInsertRepository { return getRuleNodeEventSetter(events); case DEBUG_RULE_CHAIN: return getRuleChainEventSetter(events); + case DEBUG_CALCULATED_FIELD: + return getCalculatedFieldEventSetter(events); default: throw new RuntimeException(eventType + " support is not implemented!"); } @@ -206,6 +212,29 @@ public class EventInsertRepository { }; } + private BatchPreparedStatementSetter getCalculatedFieldEventSetter(List events) { + return new BatchPreparedStatementSetter() { + @Override + public void setValues(PreparedStatement ps, int i) throws SQLException { + CalculatedFieldDebugEvent event = (CalculatedFieldDebugEvent) events.get(i); + setCommonEventFields(ps, event); + safePutUUID(ps, 6, event.getCalculatedFieldId().getId()); + safePutUUID(ps, 7, event.getEventEntity() != null ? event.getEventEntity().getId() : null); + safePutString(ps, 8, event.getEventEntity() != null ? event.getEventEntity().getEntityType().name() : null); + safePutUUID(ps, 9, event.getMsgId()); + safePutString(ps, 10, event.getMsgType()); + safePutString(ps, 11, event.getArguments()); + safePutString(ps, 12, event.getResult()); + safePutString(ps, 13, event.getError()); + } + + @Override + public int getBatchSize() { + return events.size(); + } + }; + } + void safePutString(PreparedStatement ps, int parameterIdx, String value) throws SQLException { if (value != null) { ps.setString(parameterIdx, replaceNullChars(value)); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java index b8d5083402..e3c1b37536 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java @@ -24,6 +24,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.event.CalculatedFieldDebugEventFilter; import org.thingsboard.server.common.data.event.ErrorEventFilter; import org.thingsboard.server.common.data.event.Event; import org.thingsboard.server.common.data.event.EventFilter; @@ -72,6 +73,7 @@ public class JpaBaseEventDao implements EventDao { private final RuleChainDebugEventRepository ruleChainDebugEventRepository; private final ScheduledLogExecutorComponent logExecutor; private final StatsFactory statsFactory; + private final CalculatedFieldDebugEventRepository calculatedFieldDebugEventRepository; @Value("${sql.events.batch_size:10000}") private int batchSize; @@ -110,6 +112,7 @@ public class JpaBaseEventDao implements EventDao { repositories.put(EventType.ERROR, errorEventRepository); repositories.put(EventType.DEBUG_RULE_NODE, ruleNodeDebugEventRepository); repositories.put(EventType.DEBUG_RULE_CHAIN, ruleChainDebugEventRepository); + repositories.put(EventType.DEBUG_CALCULATED_FIELD, calculatedFieldDebugEventRepository); } @PreDestroy @@ -158,6 +161,8 @@ public class JpaBaseEventDao implements EventDao { return findEventByFilter(tenantId, entityId, (ErrorEventFilter) eventFilter, pageLink); case STATS: return findEventByFilter(tenantId, entityId, (StatisticsEventFilter) eventFilter, pageLink); + case DEBUG_CALCULATED_FIELD: + return findEventByFilter(tenantId, entityId, (CalculatedFieldDebugEventFilter) eventFilter, pageLink); default: throw new RuntimeException("Not supported event type: " + eventFilter.getEventType()); } @@ -193,6 +198,8 @@ public class JpaBaseEventDao implements EventDao { case STATS: removeEventsByFilter(tenantId, entityId, (StatisticsEventFilter) eventFilter, startTime, endTime); break; + case DEBUG_CALCULATED_FIELD: + removeEventsByFilter(tenantId, entityId, (CalculatedFieldDebugEventFilter) eventFilter, startTime, endTime); default: throw new RuntimeException("Not supported event type: " + eventFilter.getEventType()); } @@ -286,6 +293,27 @@ public class JpaBaseEventDao implements EventDao { ); } + private PageData findEventByFilter(UUID tenantId, UUID entityId, CalculatedFieldDebugEventFilter eventFilter, TimePageLink pageLink) { + parseUUID(eventFilter.getCalculatedFieldId(), "Calculated Field Id"); + parseUUID(eventFilter.getEntityId(), "Entity Id"); + parseUUID(eventFilter.getMsgId(), "Message Id"); + return DaoUtil.toPageData( + calculatedFieldDebugEventRepository.findEvents( + tenantId, + entityId, + pageLink.getStartTime(), + pageLink.getEndTime(), + eventFilter.getServer(), + UUID.fromString(eventFilter.getCalculatedFieldId()), + eventFilter.getEntityId(), + eventFilter.getEntityType(), + eventFilter.getMsgId(), + eventFilter.getMsgType(), + eventFilter.isError(), + eventFilter.getErrorStr(), + DaoUtil.toPageable(pageLink, EventEntity.eventColumnMap))); + } + private void removeEventsByFilter(UUID tenantId, UUID entityId, RuleChainDebugEventFilter eventFilter, Long startTime, Long endTime) { ruleChainDebugEventRepository.removeEvents( tenantId, @@ -360,6 +388,25 @@ public class JpaBaseEventDao implements EventDao { ); } + private void removeEventsByFilter(UUID tenantId, UUID entityId, CalculatedFieldDebugEventFilter eventFilter, Long startTime, Long endTime) { + parseUUID(eventFilter.getCalculatedFieldId(), "Calculated Field Id"); + parseUUID(eventFilter.getEntityId(), "Entity Id"); + parseUUID(eventFilter.getMsgId(), "Message Id"); + calculatedFieldDebugEventRepository.removeEvents( + tenantId, + entityId, + startTime, + endTime, + eventFilter.getServer(), + UUID.fromString(eventFilter.getCalculatedFieldId()), + eventFilter.getEntityId(), + eventFilter.getEntityType(), + eventFilter.getMsgId(), + eventFilter.getMsgType(), + eventFilter.isError(), + eventFilter.getErrorStr()); + } + @Override public List findLatestEvents(UUID tenantId, UUID entityId, EventType eventType, int limit) { return DaoUtil.convertDataList(getEventRepository(eventType).findLatestEvents(tenantId, entityId, limit)); diff --git a/dao/src/main/resources/sql/schema-entities.sql b/dao/src/main/resources/sql/schema-entities.sql index 51cc66c4ac..a6d1e8800d 100644 --- a/dao/src/main/resources/sql/schema-entities.sql +++ b/dao/src/main/resources/sql/schema-entities.sql @@ -934,6 +934,7 @@ CREATE TABLE IF NOT EXISTS calculated_field ( configuration_version int DEFAULT 0, configuration varchar(1000000), version BIGINT DEFAULT 1, + debug_settings varchar(1024), external_id UUID, CONSTRAINT calculated_field_unq_key UNIQUE (entity_id, name), CONSTRAINT calculated_field_external_id_unq_key UNIQUE (tenant_id, external_id) @@ -949,3 +950,19 @@ CREATE TABLE IF NOT EXISTS calculated_field_link ( configuration varchar(10000), CONSTRAINT fk_calculated_field_id FOREIGN KEY (calculated_field_id) REFERENCES calculated_field(id) ON DELETE CASCADE ); + +CREATE TABLE IF NOT EXISTS cf_debug_event ( + id uuid NOT NULL, + tenant_id uuid NOT NULL , + ts bigint NOT NULL, + entity_id uuid NOT NULL, + service_id varchar, + cf_id uuid NOT NULL, + e_entity_id uuid, + e_entity_type varchar, + e_msg_id uuid, + e_msg_type varchar, + e_args varchar, + e_result varchar, + e_error varchar +) PARTITION BY RANGE (ts);