From a21bb92e50d26276ac06ff16c0326cd7f12d5bf3 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Thu, 30 Oct 2025 12:19:51 +0200 Subject: [PATCH] added state proto --- ...CalculatedFieldEntityMessageProcessor.java | 11 +- ...alculatedFieldManagerMessageProcessor.java | 17 +- ...tractCalculatedFieldProcessingService.java | 41 +--- .../service/cf/CalculatedFieldCache.java | 4 +- .../cf/CalculatedFieldProcessingService.java | 2 - .../cf/DefaultCalculatedFieldCache.java | 11 +- ...faultCalculatedFieldProcessingService.java | 11 +- .../DefaultCalculatedFieldQueueService.java | 9 +- .../cf/ctx/state/CalculatedFieldCtx.java | 6 - .../single/AggIntervalEntryStatus.java | 26 +- .../EntityAggregationArgumentEntry.java | 15 ++ ...EntityAggregationCalculatedFieldState.java | 226 ++++++++++-------- .../server/utils/CalculatedFieldUtils.java | 45 +++- ...gregationCalculatedFieldConfiguration.java | 2 - .../single/interval/AggInterval.java | 4 +- .../single/interval/AggIntervalType.java | 4 +- .../single/interval/BaseAggInterval.java | 72 +++--- .../single/interval/CustomInterval.java | 5 +- .../single/interval/MinInterval.java | 28 +++ .../single/interval/Watermark.java | 15 ++ common/proto/src/main/proto/queue.proto | 9 + 21 files changed, 346 insertions(+), 217 deletions(-) create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/MinInterval.java diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java index 083af60945..69ea705c60 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -31,7 +31,6 @@ 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.ReferencedEntityKey; -import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -54,6 +53,7 @@ import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesAggregationCalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.EntityAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState; @@ -127,6 +127,9 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM if (state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState) { relatedEntitiesAggState.scheduleReevaluation(); } + if (state instanceof EntityAggregationCalculatedFieldState entityAggState) { + entityAggState.scheduleReevaluation(); + } states.put(cfId, state); } else { removeState(cfId); @@ -441,11 +444,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM GeofencingCalculatedFieldState geofencingState = (GeofencingCalculatedFieldState) state; geofencingState.updateLastDynamicArgumentsRefreshTs(); } - if (ctx.getCfType() == CalculatedFieldType.ENTITY_AGGREGATION) { - var configuration = (EntityAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); - long delayUntilIntervalEnd = configuration.getInterval().getDelayUntilIntervalEnd(); - ctx.scheduleReevaluation(delayUntilIntervalEnd, actorCtx); - } + Map arguments = fetchArguments(ctx); state.update(arguments, ctx); diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index 075918e6e8..00e1dd3f96 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java @@ -34,7 +34,6 @@ import org.thingsboard.server.common.data.audit.ActionType; 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.aggregation.single.EntityAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.aggregation.RelatedEntitiesAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.CalculatedFieldId; @@ -94,7 +93,6 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private final Map> entityIdCalculatedFields = new HashMap<>(); private final Map> entityIdCalculatedFieldLinks = new HashMap<>(); private final Map> ownerEntities = new HashMap<>(); - private final Map entityAggCalculatedFields = new HashMap<>(); private ScheduledFuture cfsReevaluationTask; private final CalculatedFieldProcessingService cfExecService; @@ -307,8 +305,10 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void onRelationChangedEvent(ComponentLifecycleMsg msg, TbCallback callback) { Function> relationAction = switch (msg.getEvent()) { - case RELATION_UPDATED -> relatedId -> (entityId, ctx, cb) -> initRelatedEntity(entityId, relatedId, ctx, cb); - case RELATION_DELETED -> relatedId -> (entityId, ctx, cb) -> deleteRelatedEntity(entityId, relatedId, ctx, cb); + case RELATION_UPDATED -> + relatedId -> (entityId, ctx, cb) -> initRelatedEntity(entityId, relatedId, ctx, cb); + case RELATION_DELETED -> + relatedId -> (entityId, ctx, cb) -> deleteRelatedEntity(entityId, relatedId, ctx, cb); default -> null; }; @@ -372,9 +372,6 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); } calculatedFields.put(cf.getId(), cfCtx); - if (cf.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration aggConfig) { - entityAggCalculatedFields.put(cf.getId(), cfCtx); - } // We use copy on write lists to safely pass the reference to another actor for the iteration. // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); @@ -406,9 +403,6 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware throw CalculatedFieldException.builder().ctx(newCfCtx).eventEntity(newCfCtx.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); } finally { calculatedFields.put(newCf.getId(), newCfCtx); - if (newCf.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration aggConfig) { - entityAggCalculatedFields.put(newCf.getId(), newCfCtx); - } List oldCfList = entityIdCalculatedFields.get(newCf.getEntityId()); List newCfList = new CopyOnWriteArrayList<>(); boolean found = false; @@ -775,9 +769,6 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); } finally { calculatedFields.put(cf.getId(), cfCtx); - if (cf.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration aggConfig) { - entityAggCalculatedFields.put(cf.getId(), cfCtx); - } // We use copy on write lists to safely pass the reference to another actor for the iteration. // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java index 2110a78f5c..48bce35b3e 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java @@ -40,7 +40,6 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; -import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.ReadTsKvQuery; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.relation.EntityRelation; @@ -62,7 +61,6 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Map.Entry; import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutionException; @@ -307,42 +305,6 @@ public abstract class AbstractCalculatedFieldProcessingService { }; } - protected Map fetchMetricsDuringInterval(EntityId entityId, AggIntervalEntry interval, CalculatedFieldCtx ctx) throws Exception { - var config = (EntityAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); - Map metricsResult = new HashMap<>(); - - for (Entry entry : config.getMetrics().entrySet()) { - String metricName = entry.getKey(); - AggMetric metric = entry.getValue(); - AggFunction function = metric.getFunction(); - - AggKeyInput input = (AggKeyInput) metric.getInput(); - String argName = input.getKey(); - Argument argument = ctx.getArguments().get(argName); - String key = argument.getRefEntityKey().getKey(); - - BaseReadTsKvQuery query = new BaseReadTsKvQuery(key, interval.getStartTs(), interval.getEndTs(), 0, 1, Aggregation.valueOf(function.name())); - log.trace("[{}][{}] Fetching timeseries for query {}", ctx.getTenantId(), entityId, query); - ListenableFuture> tsFuture = timeseriesService.findAll(ctx.getTenantId(), entityId, List.of(query)); - ListenableFuture argumentEntryFut = Futures.transform(tsFuture, timeSeries -> { - log.debug("[{}][{}] Fetched {} timeseries for query {}", ctx.getTenantId(), entityId, timeSeries == null ? 0 : timeSeries.size(), query); - if (timeSeries == null || timeSeries.isEmpty()) { - return new SingleValueArgumentEntry(); - } - return ArgumentEntry.createSingleValueArgument(timeSeries.get(0)); - }, calculatedFieldCallbackExecutor); - - // Ugly but necessary. We do not expect to often fetch data from DB. Only once per pair lifetime. - // This call happens while processing the CF pack from the queue consumer. So the timeout should be relatively low. - // Alternatively, we can fetch the state outside the actor system and push separate command to create this actor, - // but this will significantly complicate the code. - ArgumentEntry argumentEntry = argumentEntryFut.get(1, TimeUnit.MINUTES); - metricsResult.put(metricName, argumentEntry); - } - - return metricsResult; - } - protected ArgumentEntry fetchMetricDuringInterval(EntityId entityId, AggIntervalEntry interval, String metricName, CalculatedFieldCtx ctx) throws Exception { var config = (EntityAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); @@ -354,7 +316,8 @@ public abstract class AbstractCalculatedFieldProcessingService { Argument argument = ctx.getArguments().get(argName); String key = argument.getRefEntityKey().getKey(); - BaseReadTsKvQuery query = new BaseReadTsKvQuery(key, interval.getStartTs(), interval.getEndTs(), 0, 1, Aggregation.valueOf(function.name())); + long intervalMs = interval.getEndTs() - interval.getStartTs(); + BaseReadTsKvQuery query = new BaseReadTsKvQuery(key, interval.getStartTs(), interval.getEndTs(), intervalMs, 1, Aggregation.valueOf(function.name())); log.trace("[{}][{}] Fetching timeseries for query {}", ctx.getTenantId(), entityId, query); ListenableFuture> tsFuture = timeseriesService.findAll(ctx.getTenantId(), entityId, List.of(query)); ListenableFuture argumentEntryFut = Futures.transform(tsFuture, timeSeries -> { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java index 27e989de70..5842363229 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java @@ -38,7 +38,9 @@ public interface CalculatedFieldCache { List getCalculatedFieldCtxsByEntityId(EntityId entityId); - List getAggCalculatedFieldCtxsByFilter(Predicate relatedEntityFilter); + List getRelatedEntitiesAggCalculatedFieldCtxsByFilter(Predicate relatedEntityFilter); + + List getEntityAggCalculatedFieldCtxsByFilter(Predicate entityAggCfFilter); boolean hasCalculatedFields(TenantId tenantId, EntityId entityId, Predicate filter); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java index 54796e18eb..e65df09f98 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java @@ -38,8 +38,6 @@ public interface CalculatedFieldProcessingService { Map fetchArgsFromDb(TenantId tenantId, EntityId entityId, Map arguments); - Map fetchMetricsDuringInterval(EntityId entityId, AggIntervalEntry interval, CalculatedFieldCtx ctx) throws Exception; - ArgumentEntry fetchMetricDuringInterval(EntityId entityId, AggIntervalEntry interval, String argName, CalculatedFieldCtx ctx) throws Exception; void pushMsgToRuleEngine(TenantId tenantId, EntityId entityId, CalculatedFieldResult result, List cfIds, TbCallback callback); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java index e802bd7677..4f2b7a459c 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java @@ -148,7 +148,7 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { } @Override - public List getAggCalculatedFieldCtxsByFilter(Predicate relatedEntityFilter) { + public List getRelatedEntitiesAggCalculatedFieldCtxsByFilter(Predicate relatedEntityFilter) { return calculatedFields.values().stream() .filter(cf -> CalculatedFieldType.RELATED_ENTITIES_AGGREGATION.equals(cf.getType())) .map(cf -> getCalculatedFieldCtx(cf.getId())) @@ -156,6 +156,15 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { .toList(); } + @Override + public List getEntityAggCalculatedFieldCtxsByFilter(Predicate entityAggCfFilter) { + return calculatedFields.values().stream() + .filter(cf -> CalculatedFieldType.ENTITY_AGGREGATION.equals(cf.getType())) + .map(cf -> getCalculatedFieldCtx(cf.getId())) + .filter(entityAggCfFilter) + .toList(); + } + @Override public boolean hasCalculatedFields(TenantId tenantId, EntityId entityId, Predicate filter) { List entityCfs = getCalculatedFieldCtxsByEntityId(entityId); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java index 35f767fde2..c6a66d100e 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java @@ -92,8 +92,10 @@ public class DefaultCalculatedFieldProcessingService extends AbstractCalculatedF @Override public Map fetchDynamicArgsFromDb(CalculatedFieldCtx ctx, EntityId entityId) { return switch (ctx.getCfType()) { - case GEOFENCING -> resolveArgumentFutures(fetchGeofencingCalculatedFieldArguments(ctx, entityId, true, System.currentTimeMillis())); - case PROPAGATION -> resolveArgumentFutures(Map.of(PROPAGATION_CONFIG_ARGUMENT, fetchPropagationCalculatedFieldArgument(ctx, entityId))); + case GEOFENCING -> + resolveArgumentFutures(fetchGeofencingCalculatedFieldArguments(ctx, entityId, true, System.currentTimeMillis())); + case PROPAGATION -> + resolveArgumentFutures(Map.of(PROPAGATION_CONFIG_ARGUMENT, fetchPropagationCalculatedFieldArgument(ctx, entityId))); default -> Collections.emptyMap(); }; } @@ -112,11 +114,6 @@ public class DefaultCalculatedFieldProcessingService extends AbstractCalculatedF return resolveArgumentFutures(argFutures); } - @Override - public Map fetchMetricsDuringInterval(EntityId entityId, AggIntervalEntry interval, CalculatedFieldCtx ctx) throws Exception { - return super.fetchMetricsDuringInterval(entityId, interval, ctx); - } - @Override public ArgumentEntry fetchMetricDuringInterval(EntityId entityId, AggIntervalEntry interval, String metricName, CalculatedFieldCtx ctx) throws Exception { return super.fetchMetricDuringInterval(entityId, interval, metricName, ctx); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java index c7e369a862..09c6fc9c38 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java @@ -188,8 +188,13 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS } } - List cfCtxs = calculatedFieldCache.getAggCalculatedFieldCtxsByFilter(relatedEntityFilter); - for (CalculatedFieldCtx cfCtx : cfCtxs) { + List entityAggCfCtxs = calculatedFieldCache.getEntityAggCalculatedFieldCtxsByFilter(filter); + if (!entityAggCfCtxs.isEmpty()) { + return true; + } + + List relatedEntityAggCfCtxs = calculatedFieldCache.getRelatedEntitiesAggCalculatedFieldCtxsByFilter(relatedEntityFilter); + for (CalculatedFieldCtx cfCtx : relatedEntityAggCfCtxs) { if (cfCtx.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig) { RelationPathLevel relation = aggConfig.getRelation(); EntitySearchDirection inverseDirection = switch (relation.direction()) { 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 0405687b4c..c0208825a8 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 @@ -37,7 +37,6 @@ import org.thingsboard.server.common.data.cf.configuration.AlarmCalculatedFieldC 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.ArgumentsBasedCalculatedFieldConfiguration; -import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.ExpressionBasedCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.Output; import org.thingsboard.server.common.data.cf.configuration.PropagationCalculatedFieldConfiguration; @@ -46,8 +45,6 @@ import org.thingsboard.server.common.data.cf.configuration.ScheduledUpdateSuppor import org.thingsboard.server.common.data.cf.configuration.SimpleCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunctionInput; import org.thingsboard.server.common.data.cf.configuration.aggregation.RelatedEntitiesAggregationCalculatedFieldConfiguration; -import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; -import org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval.AggInterval; import org.thingsboard.server.common.data.cf.configuration.geofencing.GeofencingCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; @@ -59,7 +56,6 @@ import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileCon import org.thingsboard.server.common.data.util.CollectionsUtil; import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.relation.RelationService; -import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; @@ -98,8 +94,6 @@ public class CalculatedFieldCtx implements Closeable { private String expression; private boolean useLatestTs; private boolean requiresScheduledReevaluation; -// -// private long lastReevaluationTs; private ActorSystemContext systemContext; private TbelInvokeService tbelInvokeService; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/AggIntervalEntryStatus.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/AggIntervalEntryStatus.java index 09fa341961..fa9bbd5a54 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/AggIntervalEntryStatus.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/AggIntervalEntryStatus.java @@ -1,11 +1,28 @@ +/** + * Copyright © 2016-2025 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.service.cf.ctx.state.aggregation.single; +import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import lombok.Setter; @Data @NoArgsConstructor +@AllArgsConstructor public class AggIntervalEntryStatus { @Setter @@ -19,8 +36,13 @@ public class AggIntervalEntryStatus { public boolean shouldRecalculate(long checkInterval) { boolean intervalPassed = lastMetricsEvalTs <= System.currentTimeMillis() - checkInterval; - boolean argsUpdatedDuringInterval = lastArgsRefreshTs > lastMetricsEvalTs; - return intervalPassed && argsUpdatedDuringInterval; + boolean argsUpdatedDuringInterval = lastArgsRefreshTs > -1; + if (intervalPassed && argsUpdatedDuringInterval) { + lastMetricsEvalTs = System.currentTimeMillis(); + lastArgsRefreshTs = -1; + return true; + } + return false; } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationArgumentEntry.java index 10941ebb19..c1d1b5d807 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationArgumentEntry.java @@ -1,3 +1,18 @@ +/** + * Copyright © 2016-2025 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.service.cf.ctx.state.aggregation.single; import lombok.Data; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java index 0ae745068e..33a8003e03 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java @@ -35,6 +35,7 @@ import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; +import java.util.Comparator; import java.util.HashMap; import java.util.Map; @@ -44,15 +45,54 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt private long intervalDuration; private long watermarkDuration; private long checkInterval; - private Map metrics; + private final Map> intervals = new HashMap<>(); + private CalculatedFieldProcessingService cfProcessingService; public EntityAggregationCalculatedFieldState(EntityId entityId) { super(entityId); } + public void scheduleReevaluation() { + fillMissingIntervals(interval.getCurrentIntervalEndTs(), intervalDuration); + prepareIntervals(); + long now = System.currentTimeMillis(); + intervals.forEach((intervalEntry, argumentIntervalStatuses) -> { + if (intervalEntry.belongsToInterval(now)) { + ctx.scheduleReevaluation(interval.getDelayUntilIntervalEnd(), actorCtx); + } else { + if (intervalEntry.getEndTs() <= now) { + ctx.scheduleReevaluation(checkInterval, actorCtx); + } + } + }); + } + + private void fillMissingIntervals(long currentIntervalEndTs, long intervalDuration) { + AggIntervalEntry lastIntervalEntry = intervals.keySet().stream().max(Comparator.comparing(AggIntervalEntry::getEndTs)).orElse(null); + if (lastIntervalEntry == null) { + return; + } + + long nextStartTs = lastIntervalEntry.getEndTs(); + long nextEndTs = nextStartTs + intervalDuration; + + while (nextEndTs <= currentIntervalEndTs) { + AggIntervalEntry missingAggIntervalEntry = new AggIntervalEntry(nextStartTs, nextEndTs); + + arguments.forEach((argName, argumentEntry) -> { + var entityAggEntry = (EntityAggregationArgumentEntry) argumentEntry; + entityAggEntry.getAggIntervals().put(missingAggIntervalEntry, new AggIntervalEntryStatus()); + intervals.computeIfAbsent(missingAggIntervalEntry, i -> new HashMap<>()).put(argName, new AggIntervalEntryStatus()); + }); + + nextStartTs = nextEndTs; + nextEndTs += intervalDuration; + } + } + @Override public void setCtx(CalculatedFieldCtx ctx, TbActorRef actorCtx) { super.setCtx(ctx, actorCtx); @@ -72,67 +112,17 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt @Override public ListenableFuture performCalculation(Map updatedArgs, CalculatedFieldCtx ctx) throws Exception { + createIntervalIfNotExist(); + prepareIntervals(); long now = System.currentTimeMillis(); - AggIntervalEntry aggIntervalEntry = new AggIntervalEntry(interval.getCurrentIntervalStartTs(), interval.getCurrentIntervalEndTs()); - boolean exists = false; - for (Map.Entry entry : arguments.entrySet()) { - ArgumentEntry argumentEntry = entry.getValue(); - EntityAggregationArgumentEntry entityAggEntry = (EntityAggregationArgumentEntry) argumentEntry; - Map aggIntervals = entityAggEntry.getAggIntervals(); - exists |= aggIntervals.containsKey(aggIntervalEntry); - } - if (!exists) { - arguments.forEach((argName, argumentEntry) -> { - EntityAggregationArgumentEntry entityAggEntry = (EntityAggregationArgumentEntry) argumentEntry; - entityAggEntry.getAggIntervals().put(aggIntervalEntry, new AggIntervalEntryStatus()); - }); - ctx.scheduleReevaluation(interval.getDelayUntilIntervalEnd(), actorCtx); - } Map> results = new HashMap<>(); - for (Map.Entry entry : arguments.entrySet()) { - String argName = entry.getKey(); - ArgumentEntry argumentEntry = entry.getValue(); - - EntityAggregationArgumentEntry entityAggEntry = (EntityAggregationArgumentEntry) argumentEntry; - Map aggIntervals = entityAggEntry.getAggIntervals(); - for (Map.Entry aggInterval : aggIntervals.entrySet()) { - AggIntervalEntry intervalEntry = aggInterval.getKey(); - AggIntervalEntryStatus entryStatus = aggInterval.getValue(); - - Long startTs = intervalEntry.getStartTs(); - Long endTs = intervalEntry.getEndTs(); - if (now - endTs > watermarkDuration) { - if (entryStatus.getLastArgsRefreshTs() > entryStatus.getLastMetricsEvalTs()) { - String metricName = null; - for (Map.Entry metricEntry : metrics.entrySet()) { - if (((AggKeyInput) metricEntry.getValue().getInput()).getKey().equals(argName)) { - metricName = metricEntry.getKey(); - } - } - ArgumentEntry metric = cfProcessingService.fetchMetricDuringInterval(entityId, intervalEntry, metricName, ctx); - if (!metric.isEmpty()) { - results.computeIfAbsent(intervalEntry, i -> new HashMap<>()).put(argName, metric); - } - } - aggIntervals.remove(intervalEntry); - continue; - } else if (now - startTs >= intervalDuration) { - if (entryStatus.shouldRecalculate(checkInterval)) { - String metricName = null; - for (Map.Entry metricEntry : metrics.entrySet()) { - if (((AggKeyInput) metricEntry.getValue().getInput()).getKey().equals(argName)) { - metricName = metricEntry.getKey(); - } - } - ArgumentEntry metric = cfProcessingService.fetchMetricDuringInterval(entityId, intervalEntry, metricName, ctx); - if (!metric.isEmpty()) { - results.computeIfAbsent(intervalEntry, i -> new HashMap<>()).put(argName, metric); - } - } - } - } + for (Map.Entry> entry : intervals.entrySet()) { + AggIntervalEntry intervalEntry = entry.getKey(); + Map args = entry.getValue(); + processInterval(now, intervalEntry, args, results); } + ArrayNode result = toResult(results); if (result.isEmpty()) { return Futures.immediateFuture(TelemetryCalculatedFieldResult.EMPTY); @@ -143,48 +133,86 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt .scope(output.getScope()) .result(result) .build()); + } + + private void prepareIntervals() { + arguments.forEach((argName, entry) -> { + var argEntry = (EntityAggregationArgumentEntry) entry; + argEntry.getAggIntervals().forEach((intervalEntry, status) -> + intervals.computeIfAbsent(intervalEntry, i -> new HashMap<>()).put(argName, status) + ); + }); + } + + private void createIntervalIfNotExist() { + AggIntervalEntry currentInterval = new AggIntervalEntry(interval.getCurrentIntervalStartTs(), interval.getCurrentIntervalEndTs()); + if (intervals.containsKey(currentInterval)) { + return; + } + arguments.forEach((argName, argumentEntry) -> { + var entityAggEntry = (EntityAggregationArgumentEntry) argumentEntry; + entityAggEntry.getAggIntervals().put(currentInterval, new AggIntervalEntryStatus()); + intervals.computeIfAbsent(currentInterval, i -> new HashMap<>()).put(argName, new AggIntervalEntryStatus()); + }); + ctx.scheduleReevaluation(interval.getDelayUntilIntervalEnd(), actorCtx); + } + + private void processInterval(long now, AggIntervalEntry intervalEntry, Map args, + Map> results) throws Exception { + long startTs = intervalEntry.getStartTs(); + long endTs = intervalEntry.getEndTs(); + + if (now - endTs > watermarkDuration) { + handleExpiredInterval(intervalEntry, args, results); + intervals.remove(intervalEntry); + } else if (now - startTs >= intervalDuration) { + handleActiveInterval(intervalEntry, args, results); + } + } + + private void handleExpiredInterval(AggIntervalEntry intervalEntry, + Map args, + Map> results) throws Exception { + for (Map.Entry argStatus : args.entrySet()) { + String argName = argStatus.getKey(); + AggIntervalEntryStatus argEntryIntervalStatus = argStatus.getValue(); + if (argEntryIntervalStatus.getLastArgsRefreshTs() > argEntryIntervalStatus.getLastMetricsEvalTs()) { + processMetric(intervalEntry, argName, results); + } + } + } + + private void handleActiveInterval(AggIntervalEntry intervalEntry, + Map args, + Map> results) throws Exception { + for (Map.Entry argStatus : args.entrySet()) { + String argName = argStatus.getKey(); + AggIntervalEntryStatus argEntryIntervalStatus = argStatus.getValue(); + if (argEntryIntervalStatus.shouldRecalculate(checkInterval)) { + processMetric(intervalEntry, argName, results); + ctx.scheduleReevaluation(checkInterval, actorCtx); + } + } + } + + private void processMetric(AggIntervalEntry intervalEntry, + String argName, + Map> results) throws Exception { + String metricName = findMetricName(argName); + if (metricName != null) { + ArgumentEntry metric = cfProcessingService.fetchMetricDuringInterval(entityId, intervalEntry, metricName, ctx); + if (!metric.isEmpty()) { + results.computeIfAbsent(intervalEntry, i -> new HashMap<>()).put(metricName, metric); + } + } + } -// long now = System.currentTimeMillis(); -// AggIntervalEntry aggIntervalEntry = new AggIntervalEntry(interval.getCurrentIntervalStartTs(), interval.getCurrentIntervalEndTs(), false); -// if (!intervals.containsKey(aggIntervalEntry)) { -// intervals.put(aggIntervalEntry, new AggIntervalEntryStatus()); -// ctx.scheduleReevaluation(interval.getDelayUntilIntervalEnd(), actorCtx); -// } -// ArrayNode results = JacksonUtil.newArrayNode(); -// for (Map.Entry entry : intervals.entrySet()) { -// AggIntervalEntry intervalEntry = entry.getKey(); -// AggIntervalEntryStatus entryStatus = entry.getValue(); -// -// Long startTs = intervalEntry.getStartTs(); -// Long endTs = intervalEntry.getEndTs(); -// if (now - endTs > watermarkDuration) { -// if (entryStatus.getLastArgsRefreshTs() > entryStatus.getLastMetricsEvalTs()) { -// ArgumentEntry metric = cfProcessingService.fetchMetricDuringInterval(entityId, intervalEntry, metricName, ctx); -// ObjectNode result = fetchMetrics(intervalEntry); -// if (result != null) { -// results.add(result); -// } -// } -// intervals.remove(intervalEntry); -// continue; -// } else if (now - startTs >= intervalDuration) { -// if (entryStatus.shouldRecalculate(checkInterval)) { -// ObjectNode result = fetchMetrics(intervalEntry); -// if (result != null) { -// results.add(result); -// } -// } -// } -// } -// if (results.isEmpty()) { -// return Futures.immediateFuture(TelemetryCalculatedFieldResult.EMPTY); -// } -// Output output = ctx.getOutput(); -// return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() -// .type(output.getType()) -// .scope(output.getScope()) -// .result(results) -// .build()); + private String findMetricName(String argName) { + return metrics.entrySet().stream() + .filter(e -> ((AggKeyInput) e.getValue().getInput()).getKey().equals(argName)) + .map(Map.Entry::getKey) + .findFirst() + .orElse(null); } protected ArrayNode toResult(Map> results) { diff --git a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java index 8a3e352826..780df5b795 100644 --- a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java @@ -29,6 +29,7 @@ import org.thingsboard.server.common.util.KvProtoUtil; import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.gen.transport.TransportProtos.AlarmRuleStateProto; import org.thingsboard.server.gen.transport.TransportProtos.AlarmStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.ArgumentIntervalProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; @@ -47,6 +48,10 @@ import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesArgumentEntry; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.AggIntervalEntry; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.AggIntervalEntryStatus; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.EntityAggregationArgumentEntry; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.EntityAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmRuleState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingArgumentEntry; @@ -97,14 +102,21 @@ public class CalculatedFieldUtils { state.getArguments().forEach((argName, argEntry) -> { switch (argEntry.getType()) { - case SINGLE_VALUE -> builder.addSingleValueArguments(toSingleValueArgumentProto(argName, (SingleValueArgumentEntry) argEntry)); - case TS_ROLLING -> builder.addRollingValueArguments(toRollingArgumentProto(argName, (TsRollingArgumentEntry) argEntry)); - case GEOFENCING -> builder.addGeofencingArguments(toGeofencingArgumentProto(argName, (GeofencingArgumentEntry) argEntry)); + case SINGLE_VALUE -> + builder.addSingleValueArguments(toSingleValueArgumentProto(argName, (SingleValueArgumentEntry) argEntry)); + case TS_ROLLING -> + builder.addRollingValueArguments(toRollingArgumentProto(argName, (TsRollingArgumentEntry) argEntry)); + case GEOFENCING -> + builder.addGeofencingArguments(toGeofencingArgumentProto(argName, (GeofencingArgumentEntry) argEntry)); case RELATED_ENTITIES -> { RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry = (RelatedEntitiesArgumentEntry) argEntry; relatedEntitiesArgumentEntry.getEntityInputs() .forEach((entityId, entry) -> builder.addSingleValueArguments(toSingleValueArgumentProto(argName, (SingleValueArgumentEntry) entry))); } + case ENTITY_AGGREGATION -> { + EntityAggregationArgumentEntry entityAggregationArgumentEntry = (EntityAggregationArgumentEntry) argEntry; + entityAggregationArgumentEntry.getAggIntervals().forEach((interval, argumentStatus) -> builder.addAggregationArguments(toArgumentIntervalProto(argName, interval, argumentStatus))); + } } }); if (state instanceof AlarmCalculatedFieldState alarmState) { @@ -158,6 +170,16 @@ public class CalculatedFieldUtils { return builder.build(); } + public static ArgumentIntervalProto toArgumentIntervalProto(String argName, AggIntervalEntry intervalEntry, AggIntervalEntryStatus argumentStatus) { + return ArgumentIntervalProto.newBuilder() + .setArgName(argName) + .setStartTs(intervalEntry.getStartTs()) + .setEndTs(intervalEntry.getEndTs()) + .setLastArgsRefreshTs(argumentStatus.getLastArgsRefreshTs()) + .setLastMetricsEvalTs(argumentStatus.getLastMetricsEvalTs()) + .build(); + } + public static TsRollingArgumentProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) { TsRollingArgumentProto.Builder builder = TsRollingArgumentProto.newBuilder() .setKey(argName) @@ -204,7 +226,7 @@ public class CalculatedFieldUtils { case ALARM -> new AlarmCalculatedFieldState(id.entityId()); case PROPAGATION -> new PropagationCalculatedFieldState(id.entityId()); case RELATED_ENTITIES_AGGREGATION -> new RelatedEntitiesAggregationCalculatedFieldState(id.entityId()); - case ENTITY_AGGREGATION -> null; // todo + case ENTITY_AGGREGATION -> new EntityAggregationCalculatedFieldState(id.entityId()); }; if (state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState) { @@ -222,6 +244,21 @@ public class CalculatedFieldUtils { return relatedEntitiesAggState; } + if (state instanceof EntityAggregationCalculatedFieldState entityAggregationState) { + Map arguments = new HashMap<>(); + + proto.getAggregationArgumentsList().forEach(argProto -> { + AggIntervalEntry intervalEntry = new AggIntervalEntry(argProto.getStartTs(), argProto.getEndTs()); + AggIntervalEntryStatus intervalStatus = new AggIntervalEntryStatus(argProto.getLastArgsRefreshTs(), argProto.getLastMetricsEvalTs()); + EntityAggregationArgumentEntry argEntry = arguments.computeIfAbsent(argProto.getArgName(), name -> new EntityAggregationArgumentEntry(new HashMap<>())); + argEntry.getAggIntervals().put(intervalEntry, intervalStatus); + }); + + entityAggregationState.getArguments().putAll(arguments); + + return entityAggregationState; + } + proto.getSingleValueArgumentsList().forEach(argProto -> state.getArguments().put(argProto.getArgName(), fromSingleValueArgumentProto(argProto))); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/EntityAggregationCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/EntityAggregationCalculatedFieldConfiguration.java index 3f5fc7eb96..15b85af7bc 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/EntityAggregationCalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/EntityAggregationCalculatedFieldConfiguration.java @@ -35,10 +35,8 @@ public class EntityAggregationCalculatedFieldConfiguration implements ArgumentsB @Valid @NotEmpty private Map metrics; - private AggInterval interval; private Watermark watermark; - private Output output; @Override diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggInterval.java index 966e778240..4c21c80086 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggInterval.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggInterval.java @@ -31,7 +31,9 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo; @JsonSubTypes.Type(value = WeekSunSatInterval.class, name = "WEEK_SUN_SAT"), @JsonSubTypes.Type(value = MonthInterval.class, name = "MONTH"), @JsonSubTypes.Type(value = YearInterval.class, name = "YEAR"), - @JsonSubTypes.Type(value = CustomInterval.class, name = "CUSTOM") + @JsonSubTypes.Type(value = CustomInterval.class, name = "CUSTOM"), + + @JsonSubTypes.Type(value = MinInterval.class, name = "MIN")// todo: delete. used only for tests }) @JsonIgnoreProperties(ignoreUnknown = true) public interface AggInterval { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggIntervalType.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggIntervalType.java index 62185127ed..82cf46c6ff 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggIntervalType.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggIntervalType.java @@ -23,6 +23,8 @@ public enum AggIntervalType { WEEK_SUN_SAT, MONTH, YEAR, - CUSTOM + CUSTOM, + + MIN// todo: delete } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/BaseAggInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/BaseAggInterval.java index b878ef345a..cc207133fa 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/BaseAggInterval.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/BaseAggInterval.java @@ -33,7 +33,19 @@ public abstract class BaseAggInterval implements AggInterval { @Override public long getIntervalDurationMillis() { - return getCurrentIntervalEndTs() - getCurrentIntervalStartTs(); + return getIntervalDurationMillis(getType(), 1); + } + + public long getIntervalDurationMillis(AggIntervalType type, int multiplier) { + return switch (type) { + case MIN -> Duration.ofMinutes(multiplier).toMillis(); + case HOUR -> Duration.ofHours(multiplier).toMillis(); + case DAY -> Duration.ofDays(multiplier).toMillis(); + case WEEK, WEEK_SUN_SAT -> Duration.ofDays(7L * multiplier).toMillis(); + case MONTH -> Duration.ofDays(Math.round(30 * multiplier)).toMillis(); // average + case YEAR -> Duration.ofDays(Math.round(365 * multiplier)).toMillis(); + default -> throw new IllegalArgumentException("Unsupported type: " + type); + }; } @Override @@ -42,7 +54,11 @@ public abstract class BaseAggInterval implements AggInterval { } protected long getCurrentIntervalStartTs(AggIntervalType type, int multiplier) { - return getAlignedBoundary(type, multiplier, false).toInstant().toEpochMilli(); + ZonedDateTime now = ZonedDateTime.now(); + ZonedDateTime shiftedNow = now.minus(Duration.ofMillis(offsetMillis)); + ZonedDateTime alignedStart = getAlignedBoundary(type, multiplier, false, shiftedNow); + ZonedDateTime actualStart = alignedStart.plus(Duration.ofMillis(offsetMillis)); + return actualStart.toInstant().toEpochMilli(); } @Override @@ -51,7 +67,11 @@ public abstract class BaseAggInterval implements AggInterval { } protected long getCurrentIntervalEndTs(AggIntervalType type, int multiplier) { - return getAlignedBoundary(type, multiplier, true).toInstant().toEpochMilli(); + ZonedDateTime now = ZonedDateTime.now(); + ZonedDateTime shiftedNow = now.minus(Duration.ofMillis(offsetMillis)); + ZonedDateTime alignedEnd = getAlignedBoundary(type, multiplier, true, shiftedNow); + ZonedDateTime actualEnd = alignedEnd.plus(Duration.ofMillis(offsetMillis)); + return actualEnd.toInstant().toEpochMilli(); } @Override @@ -60,40 +80,32 @@ public abstract class BaseAggInterval implements AggInterval { } protected long getDelayUntilIntervalEnd(AggIntervalType type, int multiplier) { - ZonedDateTime now = ZonedDateTime.now(); - ZonedDateTime currentStart = getAlignedBoundary(type, multiplier, false); - ZonedDateTime nextStart = getAlignedBoundary(type, multiplier, true); - - long periodMillis = Duration.between(currentStart, nextStart).toMillis(); - - // Apply offset: this shifts the grid - long off = offsetMillis % periodMillis; - if (off < 0) off += periodMillis; - - // Compute the offset-aligned start times - ZonedDateTime offsetCurrentStart = currentStart.plus(Duration.ofMillis(off)); - ZonedDateTime offsetNextStart = offsetCurrentStart.plus(Duration.ofMillis(periodMillis)); - - // If we are already past the current offset start, move to the next - ZonedDateTime target = offsetCurrentStart.isAfter(now) ? offsetCurrentStart : offsetNextStart; - // todo fix - return Math.max(Duration.between(now, target).toMillis(), 0); + long currentIntervalEndTs = getCurrentIntervalEndTs(type, multiplier); + long now = System.currentTimeMillis(); + return currentIntervalEndTs - now; } - protected ZonedDateTime getAlignedBoundary(AggIntervalType type, int multiplier, boolean next) { - ZonedDateTime now = ZonedDateTime.now(); - + protected ZonedDateTime getAlignedBoundary(AggIntervalType type, int multiplier, boolean next, ZonedDateTime reference) { return switch (type) { - case HOUR -> alignByHour(now, multiplier, next); - case DAY -> alignByDay(now, multiplier, next); - case WEEK -> alignByWeek(now, multiplier, DayOfWeek.MONDAY, next); - case WEEK_SUN_SAT -> alignByWeek(now, multiplier, DayOfWeek.SUNDAY, next); - case MONTH -> alignByMonth(now, multiplier, next); - case YEAR -> alignByYear(now, multiplier, next); + case MIN -> alignByMin(reference, multiplier, next); + case HOUR -> alignByHour(reference, multiplier, next); + case DAY -> alignByDay(reference, multiplier, next); + case WEEK -> alignByWeek(reference, multiplier, DayOfWeek.MONDAY, next); + case WEEK_SUN_SAT -> alignByWeek(reference, multiplier, DayOfWeek.SUNDAY, next); + case MONTH -> alignByMonth(reference, multiplier, next); + case YEAR -> alignByYear(reference, multiplier, next); default -> throw new IllegalArgumentException("Unsupported type: " + type); }; } + private ZonedDateTime alignByMin(ZonedDateTime now, int multiplier, boolean next) { + ZonedDateTime startOfHour = now.truncatedTo(ChronoUnit.HOURS); + long minsSinceHour = Duration.between(startOfHour, now).toHours(); + long aligned = (minsSinceHour / multiplier) * multiplier; + if (next) aligned += multiplier; + return startOfHour.plusMinutes(aligned); + } + private ZonedDateTime alignByHour(ZonedDateTime now, int multiplier, boolean next) { ZonedDateTime startOfDay = now.truncatedTo(ChronoUnit.DAYS); long hoursSinceMidnight = Duration.between(startOfDay, now).toHours(); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/CustomInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/CustomInterval.java index d875794a18..04a9adc01b 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/CustomInterval.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/CustomInterval.java @@ -15,6 +15,9 @@ */ package org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval; +import lombok.Data; + +@Data public class CustomInterval extends BaseAggInterval { private int multiplier; // number of base units (e.g. 2 hours, 5 days) @@ -27,7 +30,7 @@ public class CustomInterval extends BaseAggInterval { @Override public long getIntervalDurationMillis() { - return getCurrentIntervalEndTs() - getCurrentIntervalStartTs(); + return getIntervalDurationMillis(internalIntervalType, multiplier); } @Override diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/MinInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/MinInterval.java new file mode 100644 index 0000000000..f84cdb9b9e --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/MinInterval.java @@ -0,0 +1,28 @@ +/** + * Copyright © 2016-2025 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.cf.configuration.aggregation.single.interval; + +import lombok.Data; + +@Data +public class MinInterval extends BaseAggInterval { + + @Override + public AggIntervalType getType() { + return AggIntervalType.HOUR; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/Watermark.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/Watermark.java index 44dd6404f6..52054a12c3 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/Watermark.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/Watermark.java @@ -1,3 +1,18 @@ +/** + * Copyright © 2016-2025 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.cf.configuration.aggregation.single.interval; import lombok.Data; diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index bfdce15dbd..fa54685d6e 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -916,6 +916,14 @@ message GeofencingArgumentProto { repeated GeofencingZoneProto zones = 2; } +message ArgumentIntervalProto { + string argName = 1; + int64 startTs = 2; + int64 endTs = 3; + int64 lastArgsRefreshTs = 4; + int64 lastMetricsEvalTs = 5; +} + message CalculatedFieldStateProto { CalculatedFieldEntityCtxIdProto id = 1; string type = 2; @@ -925,6 +933,7 @@ message CalculatedFieldStateProto { AlarmStateProto alarmState = 6; int64 lastArgsUpdateTs = 7; int64 lastMetricsEvalTs = 8; + repeated ArgumentIntervalProto aggregationArguments = 9; } //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.