Artem Dzhereleiko
2 years ago
149 changed files with 3621 additions and 1838 deletions
File diff suppressed because one or more lines are too long
File diff suppressed because it is too large
File diff suppressed because one or more lines are too long
@ -1,67 +0,0 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.ProfileEntityIdInfo; |
|||
import org.thingsboard.server.common.data.page.PageDataIterable; |
|||
import org.thingsboard.server.dao.asset.AssetService; |
|||
import org.thingsboard.server.dao.device.DeviceService; |
|||
import org.thingsboard.server.queue.util.AfterStartUp; |
|||
import org.thingsboard.server.queue.util.TbRuleEngineComponent; |
|||
import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; |
|||
|
|||
@Slf4j |
|||
@Service |
|||
@TbRuleEngineComponent |
|||
@RequiredArgsConstructor |
|||
public class DefaultCalculatedFieldInitService implements CalculatedFieldInitService { |
|||
|
|||
private final CalculatedFieldEntityProfileCache entityProfileCache; |
|||
private final AssetService assetService; |
|||
private final DeviceService deviceService; |
|||
|
|||
@Value("${calculated_fields.init_fetch_pack_size:50000}") |
|||
@Getter |
|||
private int initFetchPackSize; |
|||
|
|||
@AfterStartUp(order = AfterStartUp.CF_READ_PROFILE_ENTITIES_SERVICE) |
|||
public void initCalculatedFieldDefinitions() { |
|||
PageDataIterable<ProfileEntityIdInfo> deviceIdInfos = new PageDataIterable<>(deviceService::findProfileEntityIdInfos, initFetchPackSize); |
|||
for (ProfileEntityIdInfo idInfo : deviceIdInfos) { |
|||
log.trace("Processing device record: {}", idInfo); |
|||
try { |
|||
entityProfileCache.add(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId()); |
|||
} catch (Exception e) { |
|||
log.error("Failed to process device record: {}", idInfo, e); |
|||
} |
|||
} |
|||
PageDataIterable<ProfileEntityIdInfo> assetIdInfos = new PageDataIterable<>(assetService::findProfileEntityIdInfos, initFetchPackSize); |
|||
for (ProfileEntityIdInfo idInfo : assetIdInfos) { |
|||
log.trace("Processing asset record: {}", idInfo); |
|||
try { |
|||
entityProfileCache.add(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId()); |
|||
} catch (Exception e) { |
|||
log.error("Failed to process asset record: {}", idInfo, e); |
|||
} |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -1,95 +0,0 @@ |
|||
/** |
|||
* 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.cache; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
import org.thingsboard.server.queue.discovery.QueueKey; |
|||
import org.thingsboard.server.queue.discovery.TbApplicationEventListener; |
|||
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; |
|||
import org.thingsboard.server.queue.util.TbRuleEngineComponent; |
|||
|
|||
import java.util.Collection; |
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@TbRuleEngineComponent |
|||
@Service |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
//TODO ashvayka: remove and use TenantEntityProfileCache in each CalculatedFieldManagerMessageProcessor;
|
|||
public class DefaultCalculatedFieldEntityProfileCache extends TbApplicationEventListener<PartitionChangeEvent> implements CalculatedFieldEntityProfileCache { |
|||
|
|||
private static final Integer UNKNOWN = 0; |
|||
private final ConcurrentMap<TenantId, TenantEntityProfileCache> tenantCache = new ConcurrentHashMap<>(); |
|||
private final PartitionService partitionService; |
|||
private volatile List<Integer> myPartitions = Collections.emptyList(); |
|||
|
|||
@Override |
|||
protected void onTbApplicationEvent(PartitionChangeEvent event) { |
|||
myPartitions = event.getCfPartitions().stream() |
|||
.filter(TopicPartitionInfo::isMyPartition) |
|||
.map(tpi -> tpi.getPartition().orElse(UNKNOWN)).collect(Collectors.toList()); |
|||
//Naive approach that need to be improved.
|
|||
tenantCache.values().forEach(cache -> cache.setMyPartitions(myPartitions)); |
|||
} |
|||
|
|||
@Override |
|||
public void add(TenantId tenantId, EntityId profileId, EntityId entityId) { |
|||
var tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, entityId); |
|||
var partition = tpi.getPartition().orElse(UNKNOWN); |
|||
tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()) |
|||
.add(profileId, entityId, partition, tpi.isMyPartition()); |
|||
} |
|||
|
|||
@Override |
|||
public void update(TenantId tenantId, EntityId oldProfileId, EntityId newProfileId, EntityId entityId) { |
|||
var tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, entityId); |
|||
var partition = tpi.getPartition().orElse(UNKNOWN); |
|||
var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()); |
|||
//TODO: make this method atomic;
|
|||
cache.remove(oldProfileId, entityId); |
|||
cache.add(newProfileId, entityId, partition, tpi.isMyPartition()); |
|||
} |
|||
|
|||
@Override |
|||
public void evict(TenantId tenantId, EntityId entityId) { |
|||
var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()); |
|||
cache.removeEntityId(entityId); |
|||
} |
|||
|
|||
@Override |
|||
public Collection<EntityId> getMyEntityIdsByProfileId(TenantId tenantId, EntityId profileId) { |
|||
return tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()).getMyEntityIdsByProfileId(profileId); |
|||
} |
|||
|
|||
@Override |
|||
public int getEntityIdPartition(TenantId tenantId, EntityId entityId) { |
|||
var tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, entityId); |
|||
return tpi.getPartition().orElse(UNKNOWN); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,22 @@ |
|||
/** |
|||
* 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.queue; |
|||
|
|||
public interface TbEdgeQueueAdmin extends TbQueueAdmin { |
|||
void syncEdgeNotificationsOffsets(String fatGroupId, String newGroupId); |
|||
|
|||
void deleteConsumerGroup(String consumerGroupId); |
|||
} |
|||
@ -0,0 +1,33 @@ |
|||
/** |
|||
* 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.msg.cf; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; |
|||
|
|||
@Data |
|||
public class CalculatedFieldCacheInitMsg implements ToCalculatedFieldSystemMsg { |
|||
|
|||
private final TenantId tenantId; |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.CF_CACHE_INIT_MSG; |
|||
} |
|||
|
|||
} |
|||
File diff suppressed because it is too large
@ -1,161 +0,0 @@ |
|||
{ |
|||
"annotations": { |
|||
"list": [ |
|||
{ |
|||
"builtIn": 1, |
|||
"datasource": "-- Grafana --", |
|||
"enable": true, |
|||
"hide": true, |
|||
"iconColor": "rgba(0, 211, 255, 1)", |
|||
"name": "Annotations & Alerts", |
|||
"target": { |
|||
"limit": 100, |
|||
"matchAny": false, |
|||
"tags": [], |
|||
"type": "dashboard" |
|||
}, |
|||
"type": "dashboard" |
|||
} |
|||
] |
|||
}, |
|||
"editable": true, |
|||
"fiscalYearStartMonth": 0, |
|||
"graphTooltip": 0, |
|||
"id": 6, |
|||
"iteration": 1737564772936, |
|||
"links": [], |
|||
"liveNow": false, |
|||
"panels": [ |
|||
{ |
|||
"fieldConfig": { |
|||
"defaults": { |
|||
"color": { |
|||
"mode": "palette-classic" |
|||
}, |
|||
"custom": { |
|||
"axisLabel": "", |
|||
"axisPlacement": "auto", |
|||
"barAlignment": 0, |
|||
"drawStyle": "line", |
|||
"fillOpacity": 0, |
|||
"gradientMode": "none", |
|||
"hideFrom": { |
|||
"legend": false, |
|||
"tooltip": false, |
|||
"viz": false |
|||
}, |
|||
"lineInterpolation": "smooth", |
|||
"lineWidth": 1, |
|||
"pointSize": 1, |
|||
"scaleDistribution": { |
|||
"type": "linear" |
|||
}, |
|||
"showPoints": "auto", |
|||
"spanNulls": false, |
|||
"stacking": { |
|||
"group": "A", |
|||
"mode": "none" |
|||
}, |
|||
"thresholdsStyle": { |
|||
"mode": "off" |
|||
} |
|||
}, |
|||
"decimals": 0, |
|||
"mappings": [], |
|||
"thresholds": { |
|||
"mode": "absolute", |
|||
"steps": [ |
|||
{ |
|||
"color": "green", |
|||
"value": null |
|||
}, |
|||
{ |
|||
"color": "red", |
|||
"value": 80 |
|||
} |
|||
] |
|||
} |
|||
}, |
|||
"overrides": [] |
|||
}, |
|||
"gridPos": { |
|||
"h": 10, |
|||
"w": 12, |
|||
"x": 0, |
|||
"y": 0 |
|||
}, |
|||
"id": 2, |
|||
"options": { |
|||
"legend": { |
|||
"calcs": [], |
|||
"displayMode": "list", |
|||
"placement": "bottom" |
|||
}, |
|||
"tooltip": { |
|||
"mode": "single", |
|||
"sort": "none" |
|||
} |
|||
}, |
|||
"targets": [ |
|||
{ |
|||
"datasource": { |
|||
"type": "prometheus", |
|||
"uid": "9BonzvTSz" |
|||
}, |
|||
"exemplar": true, |
|||
"expr": "sum by (objectType) (edqs_object_count{tenantId=~\"$tenantId\"})", |
|||
"interval": "", |
|||
"legendFormat": "{{objectType}}", |
|||
"refId": "A" |
|||
} |
|||
], |
|||
"title": "EDQS object count", |
|||
"type": "timeseries" |
|||
} |
|||
], |
|||
"refresh": "", |
|||
"schemaVersion": 35, |
|||
"style": "dark", |
|||
"tags": [], |
|||
"templating": { |
|||
"list": [ |
|||
{ |
|||
"current": { |
|||
"selected": true, |
|||
"text": [ |
|||
"All" |
|||
], |
|||
"value": [ |
|||
"$__all" |
|||
] |
|||
}, |
|||
"definition": "label_values(edqs_object_count, tenantId)", |
|||
"hide": 0, |
|||
"includeAll": true, |
|||
"label": "Tenant", |
|||
"multi": true, |
|||
"name": "tenantId", |
|||
"options": [], |
|||
"query": { |
|||
"query": "label_values(edqs_object_count, tenantId)", |
|||
"refId": "StandardVariableQuery" |
|||
}, |
|||
"refresh": 1, |
|||
"regex": "", |
|||
"skipUrlSync": false, |
|||
"sort": 0, |
|||
"type": "query" |
|||
} |
|||
] |
|||
}, |
|||
"time": { |
|||
"from": "now-15m", |
|||
"to": "now" |
|||
}, |
|||
"timepicker": {}, |
|||
"timezone": "", |
|||
"title": "EDQS", |
|||
"uid": "mK5A_DdHk", |
|||
"version": 9, |
|||
"weekStart": "" |
|||
} |
|||
@ -0,0 +1,127 @@ |
|||
/** |
|||
* 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.rule.engine.profile; |
|||
|
|||
import org.assertj.core.api.Assertions; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.params.ParameterizedTest; |
|||
import org.junit.jupiter.params.provider.EnumSource; |
|||
import org.junit.jupiter.params.provider.NullSource; |
|||
import org.thingsboard.server.common.data.device.profile.AlarmConditionFilterKey; |
|||
import org.thingsboard.server.common.data.device.profile.AlarmConditionKeyType; |
|||
import org.thingsboard.server.common.data.device.profile.SpecificTimeSchedule; |
|||
import org.thingsboard.server.common.data.query.ComplexFilterPredicate; |
|||
import org.thingsboard.server.common.data.query.DynamicValue; |
|||
import org.thingsboard.server.common.data.query.DynamicValueSourceType; |
|||
import org.thingsboard.server.common.data.query.FilterPredicateType; |
|||
import org.thingsboard.server.common.data.query.SimpleKeyFilterPredicate; |
|||
|
|||
import java.util.HashSet; |
|||
import java.util.Set; |
|||
|
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.BDDMockito.willCallRealMethod; |
|||
import static org.mockito.BDDMockito.willReturn; |
|||
import static org.mockito.Mockito.RETURNS_DEEP_STUBS; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
class ProfileStateTest { |
|||
|
|||
ProfileState profileState; |
|||
Set<AlarmConditionFilterKey> entityKeys = new HashSet<>(); |
|||
Set<AlarmConditionFilterKey> ruleKeys = new HashSet<>(); |
|||
|
|||
@BeforeEach |
|||
void setUp() { |
|||
profileState = mock(ProfileState.class); |
|||
} |
|||
|
|||
@ParameterizedTest() |
|||
@EnumSource(DynamicValueSourceType.class) |
|||
@NullSource |
|||
void addScheduleDynamicValuesSourceAttribute(DynamicValueSourceType sourceType) { |
|||
willCallRealMethod().given(profileState).addScheduleDynamicValues(any(), any()); |
|||
final DynamicValue<String> dynamicValue = new DynamicValue<>(sourceType, "myKey"); |
|||
SpecificTimeSchedule schedule = new SpecificTimeSchedule(); |
|||
schedule.setDynamicValue(dynamicValue); |
|||
|
|||
Assertions.assertThat(entityKeys.isEmpty()).isTrue(); |
|||
Assertions.assertThat(schedule.getDynamicValue().getSourceAttribute()).isNotNull(); |
|||
|
|||
profileState.addScheduleDynamicValues(schedule, entityKeys); |
|||
|
|||
Assertions.assertThat(entityKeys).isEqualTo(Set.of( |
|||
new AlarmConditionFilterKey(AlarmConditionKeyType.ATTRIBUTE, "myKey"))); |
|||
} |
|||
|
|||
@ParameterizedTest() |
|||
@EnumSource(DynamicValueSourceType.class) |
|||
@NullSource |
|||
void addScheduleDynamicValuesSourceAttributeIsNull(DynamicValueSourceType sourceType) { |
|||
willCallRealMethod().given(profileState).addScheduleDynamicValues(any(), any()); |
|||
DynamicValue<String> dynamicValue = new DynamicValue<>(sourceType, null); |
|||
SpecificTimeSchedule schedule = new SpecificTimeSchedule(); |
|||
schedule.setDynamicValue(dynamicValue); |
|||
|
|||
Assertions.assertThat(entityKeys.isEmpty()).isTrue(); |
|||
Assertions.assertThat(schedule.getDynamicValue().getSourceAttribute()).isNull(); |
|||
|
|||
profileState.addScheduleDynamicValues(schedule, entityKeys); |
|||
|
|||
Assertions.assertThat(entityKeys.isEmpty()).isTrue(); |
|||
} |
|||
|
|||
@ParameterizedTest() |
|||
@EnumSource(value = FilterPredicateType.class, names = {"COMPLEX"}) |
|||
void addDynamicValuesRecursivelySourceAttributeComplexKeyFilter(FilterPredicateType predicateType) { |
|||
willCallRealMethod().given(profileState).addDynamicValuesRecursively(any(), any(), any()); |
|||
ComplexFilterPredicate predicate = mock(ComplexFilterPredicate.class, RETURNS_DEEP_STUBS); |
|||
willReturn(predicateType).given(predicate).getType(); |
|||
profileState.addDynamicValuesRecursively(predicate, entityKeys, ruleKeys); |
|||
Assertions.assertThat(entityKeys.isEmpty()).isTrue(); |
|||
Assertions.assertThat(ruleKeys.isEmpty()).isTrue(); |
|||
} |
|||
|
|||
@ParameterizedTest() |
|||
@EnumSource(value = FilterPredicateType.class, names = {"STRING", "NUMERIC", "BOOLEAN"}) |
|||
void addDynamicValuesRecursivelySourceAttributeIsNull(FilterPredicateType predicateType) { |
|||
willCallRealMethod().given(profileState).addDynamicValuesRecursively(any(), any(), any()); |
|||
SimpleKeyFilterPredicate<String> predicate = mock(SimpleKeyFilterPredicate.class, RETURNS_DEEP_STUBS); |
|||
willReturn(predicateType).given(predicate).getType(); |
|||
when(predicate.getValue().getDynamicValue().getSourceType()).thenReturn(DynamicValueSourceType.CURRENT_DEVICE); |
|||
when(predicate.getValue().getDynamicValue().getSourceAttribute()).thenReturn(null); |
|||
profileState.addDynamicValuesRecursively(predicate, entityKeys, ruleKeys); |
|||
Assertions.assertThat(entityKeys.isEmpty()).isTrue(); |
|||
Assertions.assertThat(ruleKeys.isEmpty()).isTrue(); |
|||
} |
|||
|
|||
@ParameterizedTest() |
|||
@EnumSource(value = FilterPredicateType.class, names = {"STRING", "NUMERIC", "BOOLEAN"}) |
|||
void addDynamicValuesRecursivelySourceAttributeAdded(FilterPredicateType predicateType) { |
|||
willCallRealMethod().given(profileState).addDynamicValuesRecursively(any(), any(), any()); |
|||
SimpleKeyFilterPredicate<String> predicate = mock(SimpleKeyFilterPredicate.class, RETURNS_DEEP_STUBS); |
|||
willReturn(predicateType).given(predicate).getType(); |
|||
when(predicate.getValue().getDynamicValue().getSourceType()).thenReturn(DynamicValueSourceType.CURRENT_DEVICE); |
|||
when(predicate.getValue().getDynamicValue().getSourceAttribute()).thenReturn("myKey"); |
|||
profileState.addDynamicValuesRecursively(predicate, entityKeys, ruleKeys); |
|||
Assertions.assertThat(entityKeys).isEqualTo(Set.of( |
|||
new AlarmConditionFilterKey(AlarmConditionKeyType.ATTRIBUTE, "myKey"))); |
|||
Assertions.assertThat(ruleKeys).isEqualTo(Set.of( |
|||
new AlarmConditionFilterKey(AlarmConditionKeyType.ATTRIBUTE, "myKey"))); |
|||
} |
|||
|
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue