From 58e31ceb78075e591e6410b23f43c7180d3ef3e6 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Thu, 11 Feb 2021 18:19:40 +0200 Subject: [PATCH 1/8] Improvement to the restarts of the rule nodes --- .../RuleChainActorMessageProcessor.java | 3 +- .../actors/ruleChain/RuleNodeActor.java | 2 +- .../RuleNodeActorMessageProcessor.java | 9 ++- .../server/actors/service/ComponentActor.java | 4 + .../server/actors/TbActorMailbox.java | 20 ++++- .../actors/TbRuleNodeUpdateException.java | 26 +++++++ .../server/common/msg/MsgType.java | 5 ++ .../common/msg/plugin/RuleNodeUpdatedMsg.java | 40 ++++++++++ .../engine/metadata/CalculateDeltaNode.java | 75 ++++++++++--------- 9 files changed, 142 insertions(+), 42 deletions(-) create mode 100644 common/actor/src/main/java/org/thingsboard/server/actors/TbRuleNodeUpdateException.java create mode 100644 common/message/src/main/java/org/thingsboard/server/common/msg/plugin/RuleNodeUpdatedMsg.java diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java index cc0c9da8f2..830714b431 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java @@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleNode; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; +import org.thingsboard.server.common.msg.plugin.RuleNodeUpdatedMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; import org.thingsboard.server.common.msg.queue.RuleEngineException; @@ -131,7 +132,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor 0 && attemptIdx > settings.getMaxActorInitAttempts())) { log.info("[{}] Failed to init actor, attempt {}, going to stop attempts.", selfId, attempt, t); stopReason = TbActorStopReason.INIT_FAILED; - system.stop(selfId); + destroy(); } else if (strategy.getRetryDelay() > 0) { log.info("[{}] Failed to init actor, attempt {}, going to retry in attempts in {}ms", selfId, attempt, strategy.getRetryDelay()); log.debug("[{}] Error", selfId, t); @@ -95,7 +96,19 @@ public final class TbActorMailbox implements TbActorCtx { } tryProcessQueue(true); } else { - msg.onTbActorStopped(stopReason); + if (highPriority && msg.getMsgType().equals(MsgType.RULE_NODE_UPDATED_MSG)) { + synchronized (this) { + if (stopReason == TbActorStopReason.INIT_FAILED) { + destroyInProgress.set(false); + stopReason = null; + initActor(); + } else { + msg.onTbActorStopped(stopReason); + } + } + } else { + msg.onTbActorStopped(stopReason); + } } } @@ -126,6 +139,9 @@ public final class TbActorMailbox implements TbActorCtx { try { log.debug("[{}] Going to process message: {}", selfId, msg); actor.process(msg); + } catch (TbRuleNodeUpdateException updateException){ + stopReason = TbActorStopReason.INIT_FAILED; + destroy(); } catch (Throwable t) { log.debug("[{}] Failed to process message: {}", selfId, msg, t); ProcessFailureStrategy strategy = actor.onProcessFailure(t); diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbRuleNodeUpdateException.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbRuleNodeUpdateException.java new file mode 100644 index 0000000000..7e3cca863d --- /dev/null +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbRuleNodeUpdateException.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2021 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.actors; + +public class TbRuleNodeUpdateException extends RuntimeException { + + private static final long serialVersionUID = 8209771144711980882L; + + public TbRuleNodeUpdateException(String message, Throwable cause) { + super(message, cause); + } +} + diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java index ffc574d5c2..13d0a6e9eb 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java @@ -39,6 +39,11 @@ public enum MsgType { */ COMPONENT_LIFE_CYCLE_MSG, + /** + * Special message to indicate rule node update request + */ + RULE_NODE_UPDATED_MSG, + /** * Misc messages consumed from the Queue and forwarded to Rule Engine Actor. * diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/RuleNodeUpdatedMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/RuleNodeUpdatedMsg.java new file mode 100644 index 0000000000..ca4ab1c949 --- /dev/null +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/RuleNodeUpdatedMsg.java @@ -0,0 +1,40 @@ +/** + * Copyright © 2016-2021 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.msg.plugin; + +import lombok.ToString; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; +import org.thingsboard.server.common.msg.MsgType; + +import java.util.Optional; + +/** + * @author Andrew Shvayka + */ +@ToString +public class RuleNodeUpdatedMsg extends ComponentLifecycleMsg { + + public RuleNodeUpdatedMsg(TenantId tenantId, EntityId entityId) { + super(tenantId, entityId, ComponentLifecycleEvent.UPDATED); + } + + @Override + public MsgType getMsgType() { + return MsgType.RULE_NODE_UPDATED_MSG; + } +} \ No newline at end of file diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java index 0cd8a8b08a..1d5ae744c3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java @@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.util.mapping.JacksonUtil; @@ -72,41 +73,45 @@ public class CalculateDeltaNode implements TbNode { @Override public void onMsg(TbContext ctx, TbMsg msg) { - JsonNode json = JacksonUtil.toJsonNode(msg.getData()); - String inputKey = config.getInputValueKey(); - if (json.has(inputKey)) { - DonAsynchron.withCallback(getLastValue(msg.getOriginator()), - previousData -> { - double currentValue = json.get(inputKey).asDouble(); - long currentTs = TbMsgTimeseriesNode.getTs(msg); - - if (useCache) { - cache.put(msg.getOriginator(), new ValueWithTs(currentTs, currentValue)); - } - - BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0); - - if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) { - ctx.tellNext(msg, TbRelationTypes.FAILURE); - return; - } - - if (config.getRound() != null) { - delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP); - } - - ObjectNode result = (ObjectNode) json; - result.put(config.getOutputValueKey(), delta); - - if (config.isAddPeriodBetweenMsgs()) { - long period = previousData != null ? currentTs - previousData.ts : 0; - result.put(config.getPeriodValueKey(), period); - } - ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result))); - }, - t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); - } else if (config.isTellFailureIfInputValueKeyIsAbsent()) { - ctx.tellNext(msg, TbRelationTypes.FAILURE); + if (msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())) { + JsonNode json = JacksonUtil.toJsonNode(msg.getData()); + String inputKey = config.getInputValueKey(); + if (json.has(inputKey)) { + DonAsynchron.withCallback(getLastValue(msg.getOriginator()), + previousData -> { + double currentValue = json.get(inputKey).asDouble(); + long currentTs = TbMsgTimeseriesNode.getTs(msg); + + if (useCache) { + cache.put(msg.getOriginator(), new ValueWithTs(currentTs, currentValue)); + } + + BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0); + + if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) { + ctx.tellNext(msg, TbRelationTypes.FAILURE); + return; + } + + if (config.getRound() != null) { + delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP); + } + + ObjectNode result = (ObjectNode) json; + result.put(config.getOutputValueKey(), delta); + + if (config.isAddPeriodBetweenMsgs()) { + long period = previousData != null ? currentTs - previousData.ts : 0; + result.put(config.getPeriodValueKey(), period); + } + ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result))); + }, + t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); + } else if (config.isTellFailureIfInputValueKeyIsAbsent()) { + ctx.tellNext(msg, TbRelationTypes.FAILURE); + } else { + ctx.tellSuccess(msg); + } } else { ctx.tellSuccess(msg); } From 305e656c7165287740fec9fa3f28e70ab1fbd8a9 Mon Sep 17 00:00:00 2001 From: Igor Kulikov Date: Fri, 12 Feb 2021 13:52:25 +0200 Subject: [PATCH 2/8] Fix memory leak in entity data subscription service --- .../subscription/DefaultTbEntityDataSubscriptionService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java index fe2fbce573..689f80f757 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java @@ -476,7 +476,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc public void cancelAllSessionSubscriptions(String sessionId) { Map sessionSubs = subscriptionsBySessionId.remove(sessionId); if (sessionSubs != null) { - sessionSubs.values().stream().filter(sub -> sub instanceof TbEntityDataSubCtx).map(sub -> (TbEntityDataSubCtx) sub).forEach(this::cleanupAndCancel); + sessionSubs.values().forEach(this::cleanupAndCancel); } } From d34d38613b5cb91976de958db8cc0ab215603a9a Mon Sep 17 00:00:00 2001 From: Viacheslav Kukhtyn Date: Thu, 11 Feb 2021 11:57:15 +0200 Subject: [PATCH 3/8] Process alarm rules on activity and inactivity events --- .../rule/engine/profile/DeviceState.java | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java index 7ddfbce204..a4d2a2269f 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java @@ -138,6 +138,8 @@ class DeviceState { stateChanged = processTelemetry(ctx, msg); } else if (msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name())) { stateChanged = processAttributesUpdateRequest(ctx, msg); + } else if (msg.getType().equals(DataConstants.ACTIVITY_EVENT) || msg.getType().equals(DataConstants.INACTIVITY_EVENT)) { + stateChanged = processDeviceActivityEvent(ctx, msg); } else if (msg.getType().equals(DataConstants.ATTRIBUTES_UPDATED)) { stateChanged = processAttributesUpdateNotification(ctx, msg); } else if (msg.getType().equals(DataConstants.ATTRIBUTES_DELETED)) { @@ -158,6 +160,11 @@ class DeviceState { } } + private boolean processDeviceActivityEvent(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { + //TODO: Add handling a case when device state is saved in telemetry + return processAttributesUpdate(ctx, msg, msg.getMetaData().getValue("scope")); + } + private boolean processAlarmClearNotification(TbContext ctx, TbMsg msg) { boolean stateChanged = false; Alarm alarmNf = JacksonUtil.fromString(msg.getData(), Alarm.class); @@ -181,12 +188,11 @@ class DeviceState { } private boolean processAttributesUpdateNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { - Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); String scope = msg.getMetaData().getValue("scope"); if (StringUtils.isEmpty(scope)) { scope = DataConstants.CLIENT_SCOPE; } - return processAttributesUpdate(ctx, msg, attributes, scope); + return processAttributesUpdate(ctx, msg, scope); } private boolean processAttributesDeleteNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { @@ -211,12 +217,12 @@ class DeviceState { } protected boolean processAttributesUpdateRequest(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { - Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); - return processAttributesUpdate(ctx, msg, attributes, DataConstants.CLIENT_SCOPE); + return processAttributesUpdate(ctx, msg, DataConstants.CLIENT_SCOPE); } - private boolean processAttributesUpdate(TbContext ctx, TbMsg msg, Set attributes, String scope) throws ExecutionException, InterruptedException { + private boolean processAttributesUpdate(TbContext ctx, TbMsg msg, String scope) throws ExecutionException, InterruptedException { boolean stateChanged = false; + Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); if (!attributes.isEmpty()) { SnapshotUpdate update = merge(latestValues, attributes, scope); for (DeviceProfileAlarm alarm : deviceProfile.getAlarmSettings()) { From c4b1f9ea7a43847a1d78531fb96ddc78b1505855 Mon Sep 17 00:00:00 2001 From: Viacheslav Kukhtyn Date: Fri, 12 Feb 2021 12:56:16 +0200 Subject: [PATCH 4/8] Process alarms on activity and inactivity events when device state is persisted to telemetry --- .../service/state/DefaultDeviceStateService.java | 2 ++ .../server/common/data/DataConstants.java | 1 + .../rule/engine/profile/DeviceState.java | 13 ++++++++----- 3 files changed, 11 insertions(+), 5 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index b1d47c9864..e780241d3f 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java @@ -207,6 +207,7 @@ public class DefaultDeviceStateService implements DeviceStateService { state.setActive(true); save(deviceId, ACTIVITY_STATE, state.isActive()); stateData.getMetaData().putValue("scope", SERVER_SCOPE); + stateData.getMetaData().putValue(DataConstants.PERSIST_STATE_TO_TELEMETRY, String.valueOf(persistToTelemetry)); pushRuleEngineMessage(stateData, ACTIVITY_EVENT); } } @@ -385,6 +386,7 @@ public class DefaultDeviceStateService implements DeviceStateService { state.setActive(ts < state.getLastActivityTime() + state.getInactivityTimeout()); if (!state.isActive() && (state.getLastInactivityAlarmTime() == 0L || state.getLastInactivityAlarmTime() < state.getLastActivityTime()) && stateData.getDeviceCreationTime() + state.getInactivityTimeout() < ts) { state.setLastInactivityAlarmTime(ts); + stateData.getMetaData().putValue(DataConstants.PERSIST_STATE_TO_TELEMETRY, String.valueOf(persistToTelemetry)); pushRuleEngineMessage(stateData, INACTIVITY_EVENT); save(deviceId, INACTIVITY_ALARM_TIME, ts); save(deviceId, ACTIVITY_STATE, state.isActive()); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java index 12cc17270c..f238ad25e7 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java @@ -51,6 +51,7 @@ public class DataConstants { public static final String CONNECT_EVENT = "CONNECT_EVENT"; public static final String DISCONNECT_EVENT = "DISCONNECT_EVENT"; public static final String ACTIVITY_EVENT = "ACTIVITY_EVENT"; + public static final String PERSIST_STATE_TO_TELEMETRY = "persistStateToTelemetry"; public static final String ENTITY_CREATED = "ENTITY_CREATED"; public static final String ENTITY_UPDATED = "ENTITY_UPDATED"; diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java index a4d2a2269f..b5f0674993 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java @@ -161,8 +161,11 @@ class DeviceState { } private boolean processDeviceActivityEvent(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { - //TODO: Add handling a case when device state is saved in telemetry - return processAttributesUpdate(ctx, msg, msg.getMetaData().getValue("scope")); + String deviceStateIsPersistedToTelemetry = msg.getMetaData().getValue(DataConstants.PERSIST_STATE_TO_TELEMETRY); + if (Boolean.TRUE.toString().equals(deviceStateIsPersistedToTelemetry)) { + return processTelemetry(ctx, msg); + } + return processAttributes(ctx, msg, msg.getMetaData().getValue("scope")); } private boolean processAlarmClearNotification(TbContext ctx, TbMsg msg) { @@ -192,7 +195,7 @@ class DeviceState { if (StringUtils.isEmpty(scope)) { scope = DataConstants.CLIENT_SCOPE; } - return processAttributesUpdate(ctx, msg, scope); + return processAttributes(ctx, msg, scope); } private boolean processAttributesDeleteNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { @@ -217,10 +220,10 @@ class DeviceState { } protected boolean processAttributesUpdateRequest(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { - return processAttributesUpdate(ctx, msg, DataConstants.CLIENT_SCOPE); + return processAttributes(ctx, msg, DataConstants.CLIENT_SCOPE); } - private boolean processAttributesUpdate(TbContext ctx, TbMsg msg, String scope) throws ExecutionException, InterruptedException { + private boolean processAttributes(TbContext ctx, TbMsg msg, String scope) throws ExecutionException, InterruptedException { boolean stateChanged = false; Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); if (!attributes.isEmpty()) { From dac7c5250f11dde951e1ba100c8e62ec4abfd0f8 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Fri, 12 Feb 2021 15:30:11 +0200 Subject: [PATCH 5/8] Improvements to the DeviceStateService --- .../thingsboard/server/controller/BaseController.java | 4 ++-- .../service/state/DefaultDeviceStateService.java | 11 ++++++----- .../thingsboard/server/common/data/DataConstants.java | 2 +- .../action/TbCopyAttributesToEntityViewNode.java | 2 +- .../thingsboard/rule/engine/profile/DeviceState.java | 11 ++++++----- 5 files changed, 16 insertions(+), 14 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/controller/BaseController.java b/application/src/main/java/org/thingsboard/server/controller/BaseController.java index 3a315be459..6e2bffa366 100644 --- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java +++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java @@ -762,7 +762,7 @@ public abstract class BaseController { String scope = extractParameter(String.class, 0, additionalInfo); @SuppressWarnings("unchecked") List attributes = extractParameter(List.class, 1, additionalInfo); - metaData.putValue("scope", scope); + metaData.putValue(DataConstants.SCOPE, scope); if (attributes != null) { for (AttributeKvEntry attr : attributes) { addKvEntry(entityNode, attr); @@ -772,7 +772,7 @@ public abstract class BaseController { String scope = extractParameter(String.class, 0, additionalInfo); @SuppressWarnings("unchecked") List keys = extractParameter(List.class, 1, additionalInfo); - metaData.putValue("scope", scope); + metaData.putValue(DataConstants.SCOPE, scope); ArrayNode attrsArrayNode = entityNode.putArray("attributes"); if (keys != null) { keys.forEach(attrsArrayNode::add); diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index e780241d3f..3087c68f88 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java @@ -206,8 +206,6 @@ public class DefaultDeviceStateService implements DeviceStateService { if (!state.isActive()) { state.setActive(true); save(deviceId, ACTIVITY_STATE, state.isActive()); - stateData.getMetaData().putValue("scope", SERVER_SCOPE); - stateData.getMetaData().putValue(DataConstants.PERSIST_STATE_TO_TELEMETRY, String.valueOf(persistToTelemetry)); pushRuleEngineMessage(stateData, ACTIVITY_EVENT); } } @@ -386,7 +384,6 @@ public class DefaultDeviceStateService implements DeviceStateService { state.setActive(ts < state.getLastActivityTime() + state.getInactivityTimeout()); if (!state.isActive() && (state.getLastInactivityAlarmTime() == 0L || state.getLastInactivityAlarmTime() < state.getLastActivityTime()) && stateData.getDeviceCreationTime() + state.getInactivityTimeout() < ts) { state.setLastInactivityAlarmTime(ts); - stateData.getMetaData().putValue(DataConstants.PERSIST_STATE_TO_TELEMETRY, String.valueOf(persistToTelemetry)); pushRuleEngineMessage(stateData, INACTIVITY_EVENT); save(deviceId, INACTIVITY_ALARM_TIME, ts); save(deviceId, ACTIVITY_STATE, state.isActive()); @@ -449,7 +446,7 @@ public class DefaultDeviceStateService implements DeviceStateService { } private Function, DeviceStateData> extractDeviceStateData(Device device) { - return new Function, DeviceStateData>() { + return new Function<>() { @Nullable @Override public DeviceStateData apply(@Nullable List data) { @@ -505,7 +502,11 @@ public class DefaultDeviceStateService implements DeviceStateService { } else { data = JacksonUtil.toString(state); } - TbMsg tbMsg = TbMsg.newMsg(msgType, stateData.getDeviceId(), stateData.getMetaData().copy(), TbMsgDataType.JSON, data); + TbMsgMetaData md = stateData.getMetaData().copy(); + if(!persistToTelemetry){ + md.putValue(DataConstants.SCOPE, SERVER_SCOPE); + } + TbMsg tbMsg = TbMsg.newMsg(msgType, stateData.getDeviceId(), md, TbMsgDataType.JSON, data); clusterService.pushMsgToRuleEngine(stateData.getTenantId(), stateData.getDeviceId(), tbMsg, null); } catch (Exception e) { log.warn("[{}] Failed to push inactivity alarm: {}", stateData.getDeviceId(), state, e); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java index f238ad25e7..cc78e2cb45 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java @@ -24,6 +24,7 @@ public class DataConstants { public static final String CUSTOMER = "CUSTOMER"; public static final String DEVICE = "DEVICE"; + public static final String SCOPE = "scope"; public static final String CLIENT_SCOPE = "CLIENT_SCOPE"; public static final String SERVER_SCOPE = "SERVER_SCOPE"; public static final String SHARED_SCOPE = "SHARED_SCOPE"; @@ -51,7 +52,6 @@ public class DataConstants { public static final String CONNECT_EVENT = "CONNECT_EVENT"; public static final String DISCONNECT_EVENT = "DISCONNECT_EVENT"; public static final String ACTIVITY_EVENT = "ACTIVITY_EVENT"; - public static final String PERSIST_STATE_TO_TELEMETRY = "persistStateToTelemetry"; public static final String ENTITY_CREATED = "ENTITY_CREATED"; public static final String ENTITY_UPDATED = "ENTITY_UPDATED"; diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java index db95a5986c..152f8fbe49 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java @@ -76,7 +76,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { if (!msg.getMetaData().getData().isEmpty()) { long now = System.currentTimeMillis(); String scope = msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()) ? - DataConstants.CLIENT_SCOPE : msg.getMetaData().getValue("scope"); + DataConstants.CLIENT_SCOPE : msg.getMetaData().getValue(DataConstants.SCOPE); ListenableFuture> entityViewsFuture = ctx.getEntityViewService().findEntityViewsByTenantIdAndEntityIdAsync(ctx.getTenantId(), msg.getOriginator()); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java index b5f0674993..bb7d5f8862 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java @@ -161,11 +161,12 @@ class DeviceState { } private boolean processDeviceActivityEvent(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { - String deviceStateIsPersistedToTelemetry = msg.getMetaData().getValue(DataConstants.PERSIST_STATE_TO_TELEMETRY); - if (Boolean.TRUE.toString().equals(deviceStateIsPersistedToTelemetry)) { + String scope = msg.getMetaData().getValue(DataConstants.SCOPE); + if (StringUtils.isEmpty(scope)) { return processTelemetry(ctx, msg); + } else { + return processAttributes(ctx, msg, scope); } - return processAttributes(ctx, msg, msg.getMetaData().getValue("scope")); } private boolean processAlarmClearNotification(TbContext ctx, TbMsg msg) { @@ -191,7 +192,7 @@ class DeviceState { } private boolean processAttributesUpdateNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { - String scope = msg.getMetaData().getValue("scope"); + String scope = msg.getMetaData().getValue(DataConstants.SCOPE); if (StringUtils.isEmpty(scope)) { scope = DataConstants.CLIENT_SCOPE; } @@ -202,7 +203,7 @@ class DeviceState { boolean stateChanged = false; List keys = new ArrayList<>(); new JsonParser().parse(msg.getData()).getAsJsonObject().get("attributes").getAsJsonArray().forEach(e -> keys.add(e.getAsString())); - String scope = msg.getMetaData().getValue("scope"); + String scope = msg.getMetaData().getValue(DataConstants.SCOPE); if (StringUtils.isEmpty(scope)) { scope = DataConstants.CLIENT_SCOPE; } From eaa2c5785f3b5e72db1592511cf2fedc355067c4 Mon Sep 17 00:00:00 2001 From: VoBa Date: Fri, 12 Feb 2021 16:06:01 +0200 Subject: [PATCH 6/8] Handle case when device was removed from db but message in the queue (#4092) * Remove device from cache in case null value cached in the distributed redis * Handle case when device was removed from db but message in the queue exists * Code review chagnes --- .../thingsboard/rule/engine/profile/TbDeviceProfileNode.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java index 3faac79420..9b891ea8eb 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java @@ -134,7 +134,8 @@ public class TbDeviceProfileNode implements TbNode { if (deviceState != null) { deviceState.process(ctx, msg); } else { - ctx.tellFailure(msg, new IllegalStateException("Device profile for device [" + deviceId + "] not found!")); + log.info("Device was not found! Most probably device [" + deviceId + "] has been removed from the database. Acknowledging msg."); + ctx.ack(msg); } } } else { From 411c9dabdafa687362e66e7b3358405f1431fe4c Mon Sep 17 00:00:00 2001 From: VoBa Date: Mon, 15 Feb 2021 12:24:30 +0200 Subject: [PATCH 7/8] Added usage statistics configuration to yml file (#4097) * Remove device from cache in case null value cached in the distributed redis * Handle case when device was removed from db but message in the queue exists * Code review chagnes * Added usage statistics configuration to yml file --- application/src/main/resources/thingsboard.yml | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 0cc7c08669..4e9693448a 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -118,6 +118,15 @@ security: githubMapper: emailUrl: "${SECURITY_OAUTH2_GITHUB_MAPPER_EMAIL_URL_KEY:https://api.github.com/user/emails}" +# Usage statistics parameters +usage: + stats: + report: + enabled: "${USAGE_STATS_REPORT_ENABLED:true}" + interval: "${USAGE_STATS_REPORT_INTERVAL:10}" + check: + cycle: "${USAGE_STATS_CHECK_CYCLE:60000}" + # Dashboard parameters dashboard: # Maximum allowed datapoints fetched by widgets From 6c1074a8b04767c652940aba3b3298f075727b89 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Mon, 15 Feb 2021 14:25:40 +0200 Subject: [PATCH 8/8] Fix for race condition in the partition change events --- .../actors/service/DefaultActorService.java | 9 ++-- .../DefaultTbApiUsageStateService.java | 5 +- .../queue/DefaultTbCoreConsumerService.java | 10 ++-- .../DefaultTbRuleEngineConsumerService.java | 10 ++-- .../processing/AbstractConsumerService.java | 3 +- .../state/DefaultDeviceStateService.java | 5 +- .../DefaultSubscriptionManagerService.java | 5 +- .../DefaultTbLocalSubscriptionService.java | 49 +++++++++++------ .../AbstractSubscriptionService.java | 6 +-- .../discovery/ClusterTopologyChangeEvent.java | 4 +- .../queue/discovery/HashPartitionService.java | 4 +- .../queue/discovery/PartitionChangeEvent.java | 4 +- .../queue/discovery/TbApplicationEvent.java | 37 +++++++++++++ .../discovery/TbApplicationEventListener.java | 52 +++++++++++++++++++ 14 files changed, 158 insertions(+), 45 deletions(-) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java diff --git a/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java b/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java index 6b7b5ff3d1..05363dfd59 100644 --- a/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java +++ b/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java @@ -34,6 +34,7 @@ import org.thingsboard.server.actors.app.AppInitMsg; import org.thingsboard.server.actors.stats.StatsActor; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; @@ -43,7 +44,7 @@ import java.util.concurrent.ScheduledExecutorService; @Service @Slf4j -public class DefaultActorService implements ActorService { +public class DefaultActorService extends TbApplicationEventListener implements ActorService { public static final String APP_DISPATCHER_NAME = "app-dispatcher"; public static final String TENANT_DISPATCHER_NAME = "tenant-dispatcher"; @@ -120,10 +121,10 @@ public class DefaultActorService implements ActorService { appActor.tellWithHighPriority(new AppInitMsg()); } - @EventListener(PartitionChangeEvent.class) - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + @Override + protected void onTbApplicationEvent(PartitionChangeEvent event) { log.info("Received partition change event."); - this.appActor.tellWithHighPriority(new PartitionChangeMsg(partitionChangeEvent.getServiceQueueKey(), partitionChangeEvent.getPartitions())); + this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceQueueKey(), event.getPartitions())); } @PreDestroy diff --git a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java b/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java index d4a5d42320..d0a3984660 100644 --- a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java @@ -54,6 +54,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.UsageStatsKVProto; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.scheduler.SchedulerComponent; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.telemetry.InternalTelemetryService; @@ -78,7 +79,7 @@ import java.util.stream.Collectors; @Slf4j @Service -public class DefaultTbApiUsageStateService implements TbApiUsageStateService { +public class DefaultTbApiUsageStateService extends TbApplicationEventListener implements TbApiUsageStateService { public static final String HOURLY = "Hourly"; public static final FutureCallback VOID_CALLBACK = new FutureCallback() { @@ -188,7 +189,7 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService { } @Override - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) { if (partitionChangeEvent.getServiceType().equals(ServiceType.TB_CORE)) { myTenantStates.entrySet().removeIf(entry -> !partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition()); otherTenantStates.entrySet().removeIf(entry -> partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition()); 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 df0e7f86b5..af9239d298 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 @@ -151,12 +151,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService tpi.newByTopic(usageStatsConsumer.getTopic())) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 106ca40c94..390798a3e2 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -140,11 +140,11 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } @Override - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { - if (partitionChangeEvent.getServiceType().equals(getServiceType())) { - ServiceQueue serviceQueue = partitionChangeEvent.getServiceQueueKey().getServiceQueue(); - log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), partitionChangeEvent.getPartitions()); - consumers.get(serviceQueue.getQueue()).subscribe(partitionChangeEvent.getPartitions()); + protected void onTbApplicationEvent(PartitionChangeEvent event) { + if (event.getServiceType().equals(getServiceType())) { + ServiceQueue serviceQueue = event.getServiceQueueKey().getServiceQueue(); + log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), event.getPartitions()); + consumers.get(serviceQueue.getQueue()).subscribe(event.getPartitions()); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 31d5cf47c3..02378eb557 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -36,6 +36,7 @@ import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; @@ -56,7 +57,7 @@ import java.util.function.Function; import java.util.stream.Collectors; @Slf4j -public abstract class AbstractConsumerService implements ApplicationListener { +public abstract class AbstractConsumerService extends TbApplicationEventListener { protected volatile ExecutorService consumersExecutor; protected volatile ExecutorService notificationsConsumerExecutor; diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index 3087c68f88..be31a0df45 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java @@ -56,6 +56,7 @@ import org.thingsboard.server.dao.util.mapping.JacksonUtil; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; @@ -90,7 +91,7 @@ import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE; @Service @TbCoreComponent @Slf4j -public class DefaultDeviceStateService implements DeviceStateService { +public class DefaultDeviceStateService extends TbApplicationEventListener implements DeviceStateService { public static final String ACTIVITY_STATE = "active"; public static final String LAST_CONNECT_TIME = "lastConnectTime"; @@ -294,7 +295,7 @@ public class DefaultDeviceStateService implements DeviceStateService { } @Override - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) { if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) { deduplicationExecutor.submit(partitionChangeEvent.getPartitions()); } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index bafbb45ba8..844db92de8 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -48,6 +48,7 @@ import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -76,7 +77,7 @@ import java.util.function.Predicate; @Slf4j @TbCoreComponent @Service -public class DefaultSubscriptionManagerService implements SubscriptionManagerService { +public class DefaultSubscriptionManagerService extends TbApplicationEventListener implements SubscriptionManagerService { @Autowired private AttributesService attrService; @@ -178,7 +179,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer } @Override - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) { if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) { Set removedPartitions = new HashSet<>(currentPartitions); removedPartitions.removeAll(partitionChangeEvent.getPartitions()); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index ee00b28562..0220a94964 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java @@ -28,6 +28,7 @@ import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate; @@ -62,6 +63,34 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer private SubscriptionManagerService subscriptionManagerService; private ExecutorService subscriptionUpdateExecutor; + + private TbApplicationEventListener partitionChangeListener = new TbApplicationEventListener<>() { + @Override + protected void onTbApplicationEvent(PartitionChangeEvent event) { + if (ServiceType.TB_CORE.equals(event.getServiceType())) { + currentPartitions.clear(); + currentPartitions.addAll(event.getPartitions()); + } + } + }; + + private TbApplicationEventListener clusterTopologyChangeListener = new TbApplicationEventListener<>() { + @Override + protected void onTbApplicationEvent(ClusterTopologyChangeEvent event) { + if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) { + /* + * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again. + * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart. + * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically + * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService. + * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache + * Since number of subscriptions is usually much less then number of devices that are pushing data. + */ + subscriptionsBySessionId.values().forEach(map -> map.values() + .forEach(sub -> pushSubscriptionToManagerService(sub, true))); + } + } + }; @PostConstruct public void initExecutor() { @@ -77,28 +106,14 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer @Override @EventListener(PartitionChangeEvent.class) - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { - if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) { - currentPartitions.clear(); - currentPartitions.addAll(partitionChangeEvent.getPartitions()); - } + public void onApplicationEvent(PartitionChangeEvent event) { + partitionChangeListener.onApplicationEvent(event); } @Override @EventListener(ClusterTopologyChangeEvent.class) public void onApplicationEvent(ClusterTopologyChangeEvent event) { - if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) { - /* - * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again. - * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart. - * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically - * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService. - * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache - * Since number of subscriptions is usually much less then number of devices that are pushing data. - */ - subscriptionsBySessionId.values().forEach(map -> map.values() - .forEach(sub -> pushSubscriptionToManagerService(sub, true))); - } + clusterTopologyChangeListener.onApplicationEvent(event); } //TODO 3.1: replace null callbacks with callbacks from websocket service. diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java index 8827a71f70..168e1271fd 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java @@ -41,6 +41,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.subscription.SubscriptionManagerService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; @@ -61,7 +62,7 @@ import java.util.function.Consumer; * Created by ashvayka on 27.03.18. */ @Slf4j -public abstract class AbstractSubscriptionService implements ApplicationListener { +public abstract class AbstractSubscriptionService extends TbApplicationEventListener{ protected final Set currentPartitions = ConcurrentHashMap.newKeySet(); @@ -97,8 +98,7 @@ public abstract class AbstractSubscriptionService implements ApplicationListener } @Override - @EventListener(PartitionChangeEvent.class) - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) { if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) { currentPartitions.clear(); currentPartitions.addAll(partitionChangeEvent.getPartitions()); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java index 0602dee3ae..1e5b90b5fe 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java @@ -22,7 +22,9 @@ import org.thingsboard.server.common.msg.queue.ServiceQueueKey; import java.util.Set; -public class ClusterTopologyChangeEvent extends ApplicationEvent { +public class ClusterTopologyChangeEvent extends TbApplicationEvent { + + private static final long serialVersionUID = -2441739930040282254L; @Getter private final Set serviceQueueKeys; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java index d8164f4be8..2da438417a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java @@ -126,7 +126,7 @@ public class HashPartitionService implements PartitionService { } @Override - public void recalculatePartitions(ServiceInfo currentService, List otherServices) { + public synchronized void recalculatePartitions(ServiceInfo currentService, List otherServices) { logServiceInfo(currentService); otherServices.forEach(this::logServiceInfo); Map> queueServicesMap = new HashMap<>(); @@ -134,7 +134,7 @@ public class HashPartitionService implements PartitionService { for (ServiceInfo other : otherServices) { addNode(queueServicesMap, other); } - queueServicesMap.values().forEach(list -> list.sort((a, b) -> a.getServiceId().compareTo(b.getServiceId()))); + queueServicesMap.values().forEach(list -> list.sort(Comparator.comparing(ServiceInfo::getServiceId))); ConcurrentMap> oldPartitions = myPartitions; TenantId myIsolatedOrSystemTenantId = getSystemOrIsolatedTenantId(currentService); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java index e4edabbe19..2edcd2ceca 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java @@ -24,7 +24,9 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import java.util.Set; -public class PartitionChangeEvent extends ApplicationEvent { +public class PartitionChangeEvent extends TbApplicationEvent { + + private static final long serialVersionUID = -8731788167026510559L; @Getter private final ServiceQueueKey serviceQueueKey; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java new file mode 100644 index 0000000000..face2d36d6 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2021 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.queue.discovery; + +import lombok.Getter; +import org.springframework.context.ApplicationEvent; + +import java.util.concurrent.atomic.AtomicInteger; + +public class TbApplicationEvent extends ApplicationEvent { + + private static final long serialVersionUID = 3884264064887765146L; + + private static final AtomicInteger sequence = new AtomicInteger(); + + @Getter + private final int sequenceNumber; + + public TbApplicationEvent(Object source) { + super(source); + sequenceNumber = sequence.incrementAndGet(); + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java new file mode 100644 index 0000000000..9158d8f0c8 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java @@ -0,0 +1,52 @@ +/** + * Copyright © 2016-2021 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.queue.discovery; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.ApplicationListener; + +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +@Slf4j +public abstract class TbApplicationEventListener implements ApplicationListener { + + private int lastProcessedSequenceNumber = Integer.MIN_VALUE; + private final Lock seqNumberLock = new ReentrantLock(); + + @Override + public void onApplicationEvent(T event) { + boolean validUpdate = false; + seqNumberLock.lock(); + try { + if (event.getSequenceNumber() > lastProcessedSequenceNumber) { + validUpdate = true; + lastProcessedSequenceNumber = event.getSequenceNumber(); + } + } finally { + seqNumberLock.unlock(); + } + if (validUpdate) { + onTbApplicationEvent(event); + } else { + log.info("Application event ignored due to invalid sequence number ({} > {}). Event: {}", lastProcessedSequenceNumber, event.getSequenceNumber(), event); + } + } + + protected abstract void onTbApplicationEvent(T event); + + +}