Browse Source

Local notification rules cache

pull/8334/head
ViacheslavKlimov 4 years ago
parent
commit
788013c64a
  1. 6
      application/src/main/java/org/thingsboard/server/controller/NotificationRuleController.java
  2. 45
      application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java
  3. 116
      application/src/main/java/org/thingsboard/server/service/notification/rule/cache/DefaultNotificationRulesCache.java
  4. 19
      application/src/main/java/org/thingsboard/server/service/notification/rule/cache/NotificationRulesCache.java
  5. 3
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
  6. 7
      application/src/main/resources/thingsboard.yml
  7. 1
      common/data/src/main/java/org/thingsboard/server/common/data/CacheConstants.java
  8. 43
      dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationRuleService.java
  9. 43
      dao/src/main/java/org/thingsboard/server/dao/notification/cache/NotificationRuleCacheKey.java
  10. 32
      dao/src/main/java/org/thingsboard/server/dao/notification/cache/NotificationRuleCaffeineCache.java
  11. 35
      dao/src/main/java/org/thingsboard/server/dao/notification/cache/NotificationRuleRedisCache.java

6
application/src/main/java/org/thingsboard/server/controller/NotificationRuleController.java

@ -67,7 +67,11 @@ public class NotificationRuleController extends BaseController {
throw new IllegalArgumentException("Trigger type " + triggerType + " is not available");
}
return doSaveAndLog(EntityType.NOTIFICATION_RULE, notificationRule, notificationRuleService::saveNotificationRule);
boolean created = notificationRule.getId() == null;
notificationRule = doSaveAndLog(EntityType.NOTIFICATION_RULE, notificationRule, notificationRuleService::saveNotificationRule);
tbClusterService.broadcastEntityStateChangeEvent(user.getTenantId(), notificationRule.getId(), created ?
ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED);
return notificationRule;
}
@GetMapping("/rule/{id}")

45
application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java

@ -35,12 +35,14 @@ import org.thingsboard.server.common.data.notification.rule.NotificationRule;
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerConfig;
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
import org.thingsboard.server.common.msg.notification.trigger.NotificationRuleTrigger;
import org.thingsboard.server.common.msg.notification.trigger.RuleEngineMsgTrigger;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.dao.notification.NotificationRequestService;
import org.thingsboard.server.dao.notification.NotificationRuleService;
import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
import org.thingsboard.server.service.notification.rule.cache.NotificationRulesCache;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.service.executors.NotificationExecutorService;
import org.thingsboard.server.service.notification.rule.trigger.NotificationRuleTriggerProcessor;
import org.thingsboard.server.service.notification.rule.trigger.RuleEngineMsgNotificationRuleTriggerProcessor;
@ -60,8 +62,9 @@ import java.util.stream.Collectors;
@SuppressWarnings({"rawtypes", "unchecked"})
public class DefaultNotificationRuleProcessor implements NotificationRuleProcessor {
private final NotificationRuleService notificationRuleService;
private final NotificationRulesCache notificationRulesCache;
private final NotificationRequestService notificationRequestService;
private final PartitionService partitionService;
@Autowired @Lazy
private NotificationCenter notificationCenter;
private final NotificationExecutorService notificationExecutor;
@ -74,15 +77,19 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
if (triggerType == null) return;
TenantId tenantId = triggerType.isTenantLevel() ? trigger.getTenantId() : TenantId.SYS_TENANT_ID;
List<NotificationRule> rules = notificationRuleService.findNotificationRulesByTenantIdAndTriggerType(tenantId, triggerType);
for (NotificationRule rule : rules) {
notificationExecutor.submit(() -> {
try {
processNotificationRule(rule, trigger);
} catch (Throwable e) {
log.error("Failed to process notification rule {} for trigger type {} with trigger object {}", rule.getId(), rule.getTriggerType(), trigger, e);
}
});
try {
List<NotificationRule> rules = notificationRulesCache.get(tenantId, triggerType);
for (NotificationRule rule : rules) {
notificationExecutor.submit(() -> {
try {
processNotificationRule(rule, trigger);
} catch (Throwable e) {
log.error("Failed to process notification rule {} for trigger type {} with trigger object {}", rule.getId(), rule.getTriggerType(), trigger, e);
}
});
}
} catch (Throwable e) {
log.error("Failed to process notification rules for trigger: {}", trigger, e);
}
}
@ -169,12 +176,14 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
TenantId tenantId = componentLifecycleMsg.getTenantId();
NotificationRuleId notificationRuleId = (NotificationRuleId) componentLifecycleMsg.getEntityId();
notificationExecutor.submit(() -> {
List<NotificationRequestId> scheduledForRule = notificationRequestService.findNotificationRequestsIdsByStatusAndRuleId(tenantId, NotificationRequestStatus.SCHEDULED, notificationRuleId);
for (NotificationRequestId notificationRequestId : scheduledForRule) {
notificationCenter.deleteNotificationRequest(tenantId, notificationRequestId);
}
});
if (partitionService.isMyPartition(ServiceType.TB_CORE, tenantId, notificationRuleId)) {
notificationExecutor.submit(() -> {
List<NotificationRequestId> scheduledForRule = notificationRequestService.findNotificationRequestsIdsByStatusAndRuleId(tenantId, NotificationRequestStatus.SCHEDULED, notificationRuleId);
for (NotificationRequestId notificationRequestId : scheduledForRule) {
notificationCenter.deleteNotificationRequest(tenantId, notificationRequestId);
}
});
}
}
@Autowired

116
application/src/main/java/org/thingsboard/server/service/notification/rule/cache/DefaultNotificationRulesCache.java

@ -0,0 +1,116 @@
/**
* Copyright © 2016-2023 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.notification.rule.cache;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.notification.rule.NotificationRule;
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.dao.notification.NotificationRuleService;
import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
@Service
@RequiredArgsConstructor
@Slf4j
public class DefaultNotificationRulesCache implements NotificationRulesCache {
private final NotificationRuleService notificationRuleService;
@Value("${cache.notificationRules.maxSize:1000}")
private int cacheMaxSize;
@Value("${cache.notificationRules.timeToLiveInMinutes:30}")
private int cacheValueTtl;
private Cache<CacheKey, List<NotificationRule>> cache;
private final ReadWriteLock lock = new ReentrantReadWriteLock();
@PostConstruct
private void init() {
cache = Caffeine.newBuilder()
.maximumSize(cacheMaxSize)
.expireAfterAccess(cacheValueTtl, TimeUnit.MINUTES)
.build();
}
@EventListener(ComponentLifecycleMsg.class)
public void onComponentLifecycleEvent(ComponentLifecycleMsg event) {
switch (event.getEntityId().getEntityType()) {
case NOTIFICATION_RULE:
evict(event.getTenantId()); // TODO: evict by trigger type of the rule
break;
case TENANT:
if (event.getEvent() == ComponentLifecycleEvent.DELETED) {
lock.writeLock().lock(); // locking in case rules for tenant are fetched while evicting
try {
evict(event.getTenantId());
} finally {
lock.writeLock().unlock();
}
}
break;
}
}
@Override
public List<NotificationRule> get(TenantId tenantId, NotificationRuleTriggerType triggerType) {
lock.readLock().lock();
try {
log.trace("Retrieving notification rules of type {} for tenant {} from cache", triggerType, tenantId);
return cache.get(key(tenantId, triggerType), k -> {
List<NotificationRule> rules = notificationRuleService.findNotificationRulesByTenantIdAndTriggerType(tenantId, triggerType);
log.trace("Fetched notification rules of type {} for tenant {} (count: {})", triggerType, tenantId, rules.size());
return !rules.isEmpty() ? rules : Collections.emptyList();
});
} finally {
lock.readLock().unlock();
}
}
private void evict(TenantId tenantId) {
cache.invalidateAll(Arrays.stream(NotificationRuleTriggerType.values())
.map(triggerType -> key(tenantId, triggerType))
.collect(Collectors.toList()));
log.trace("Evicted all notification rules for tenant {} from cache", tenantId);
}
private static CacheKey key(TenantId tenantId, NotificationRuleTriggerType triggerType) {
return new CacheKey(tenantId, triggerType);
}
@Data
private static class CacheKey {
private final TenantId tenantId;
private final NotificationRuleTriggerType triggerType;
}
}

19
dao/src/main/java/org/thingsboard/server/dao/notification/cache/NotificationRuleCacheValue.java → application/src/main/java/org/thingsboard/server/service/notification/rule/cache/NotificationRulesCache.java

@ -13,25 +13,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.dao.notification.cache;
package org.thingsboard.server.service.notification.rule.cache;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.notification.rule.NotificationRule;
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
import java.io.Serializable;
import java.util.List;
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class NotificationRuleCacheValue implements Serializable {
public interface NotificationRulesCache {
private static final long serialVersionUID = 9503216785105415L;
private List<NotificationRule> notificationRules;
List<NotificationRule> get(TenantId tenantId, NotificationRuleTriggerType triggerType);
}

3
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java

@ -414,7 +414,8 @@ public class DefaultTbClusterService implements TbClusterService {
|| entityType.equals(EntityType.API_USAGE_STATE)
|| (entityType.equals(EntityType.DEVICE) && msg.getEvent() == ComponentLifecycleEvent.UPDATED)
|| entityType.equals(EntityType.ENTITY_VIEW)
|| entityType.equals(EntityType.EDGE)) {
|| entityType.equals(EntityType.EDGE)
|| entityType.equals(EntityType.NOTIFICATION_RULE)) {
TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer();
Set<String> tbCoreServices = partitionService.getAllServiceIds(ServiceType.TB_CORE);
for (String serviceId : tbCoreServices) {

7
application/src/main/resources/thingsboard.yml

@ -437,9 +437,6 @@ cache:
assetProfiles:
timeToLiveInMinutes: "${CACHE_SPECS_ASSET_PROFILES_TTL:1440}"
maxSize: "${CACHE_SPECS_ASSET_PROFILES_MAX_SIZE:10000}"
notificationRules:
timeToLiveInMinutes: "${CACHE_SPECS_NOTIFICATION_RULES_TTL:1440}"
maxSize: "${CACHE_SPECS_NOTIFICATION_RULES_MAX_SIZE:10000}"
notificationSettings:
timeToLiveInMinutes: "${CACHE_SPECS_NOTIFICATION_SETTINGS_TTL:10}"
maxSize: "${CACHE_SPECS_NOTIFICATION_SETTINGS_MAX_SIZE:1000}"
@ -480,6 +477,10 @@ cache:
entityCount:
timeToLiveInMinutes: "${CACHE_SPECS_ENTITY_COUNT_TTL:1440}"
maxSize: "${CACHE_SPECS_ENTITY_COUNT_MAX_SIZE:100000}"
# deliberately placed outside 'specs' group above
notificationRules:
timeToLiveInMinutes: "${CACHE_SPECS_NOTIFICATION_RULES_TTL:30}"
maxSize: "${CACHE_SPECS_NOTIFICATION_RULES_MAX_SIZE:1000}"
#Disable this because it is not required.
spring.data.redis.repositories.enabled: false

1
common/data/src/main/java/org/thingsboard/server/common/data/CacheConstants.java

@ -29,7 +29,6 @@ public class CacheConstants {
public static final String TENANTS_CACHE = "tenants";
public static final String TENANTS_EXIST_CACHE = "tenantsExist";
public static final String DEVICE_PROFILE_CACHE = "deviceProfiles";
public static final String NOTIFICATION_RULES_CACHE = "notificationRules";
public static final String NOTIFICATION_SETTINGS_CACHE = "notificationSettings";
public static final String ASSET_PROFILE_CACHE = "assetProfiles";

43
dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationRuleService.java

@ -17,7 +17,6 @@ package org.thingsboard.server.dao.notification;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.HasId;
@ -28,20 +27,16 @@ import org.thingsboard.server.common.data.notification.rule.NotificationRuleInfo
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.dao.entity.AbstractCachedEntityService;
import org.thingsboard.server.dao.entity.AbstractEntityService;
import org.thingsboard.server.dao.entity.EntityDaoService;
import org.thingsboard.server.dao.notification.cache.NotificationRuleCacheKey;
import org.thingsboard.server.dao.notification.cache.NotificationRuleCacheValue;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@Service
@RequiredArgsConstructor
public class DefaultNotificationRuleService extends AbstractCachedEntityService<NotificationRuleCacheKey, NotificationRuleCacheValue, NotificationRule> implements NotificationRuleService, EntityDaoService {
public class DefaultNotificationRuleService extends AbstractEntityService implements NotificationRuleService, EntityDaoService {
private final NotificationRuleDao notificationRuleDao;
@ -55,16 +50,13 @@ public class DefaultNotificationRuleService extends AbstractCachedEntityService<
}
}
try {
notificationRule = notificationRuleDao.saveAndFlush(tenantId, notificationRule);
publishEvictEvent(notificationRule);
return notificationRuleDao.saveAndFlush(tenantId, notificationRule);
} catch (Exception e) {
handleEvictEvent(notificationRule);
checkConstraintViolation(e, Map.of(
"uq_notification_rule_name", "Notification rule with such name already exists"
));
throw e;
}
return notificationRule;
}
@Override
@ -89,44 +81,17 @@ public class DefaultNotificationRuleService extends AbstractCachedEntityService<
@Override
public List<NotificationRule> findNotificationRulesByTenantIdAndTriggerType(TenantId tenantId, NotificationRuleTriggerType triggerType) {
NotificationRuleCacheKey cacheKey = NotificationRuleCacheKey.builder()
.tenantId(tenantId)
.triggerType(triggerType)
.build();
return cache.getAndPutInTransaction(cacheKey, () -> NotificationRuleCacheValue.builder()
.notificationRules(notificationRuleDao.findByTenantIdAndTriggerType(tenantId, triggerType))
.build(), false)
.getNotificationRules();
return notificationRuleDao.findByTenantIdAndTriggerType(tenantId, triggerType);
}
@Transactional
@Override
public void deleteNotificationRuleById(TenantId tenantId, NotificationRuleId id) {
NotificationRule notificationRule = findNotificationRuleById(tenantId, id);
publishEvictEvent(notificationRule);
notificationRuleDao.removeById(tenantId, id.getId());
}
@Override
public void deleteNotificationRulesByTenantId(TenantId tenantId) {
notificationRuleDao.removeByTenantId(tenantId);
List<NotificationRuleCacheKey> cacheKeys = Arrays.stream(NotificationRuleTriggerType.values())
.map(triggerType -> NotificationRuleCacheKey.builder()
.tenantId(tenantId)
.triggerType(triggerType)
.build())
.collect(Collectors.toList());
cache.evict(cacheKeys);
}
@Override
public void handleEvictEvent(NotificationRule notificationRule) {
NotificationRuleCacheKey cacheKey = NotificationRuleCacheKey.builder()
.tenantId(notificationRule.getTenantId())
.triggerType(notificationRule.getTriggerType())
.build();
cache.evict(cacheKey);
}
@Override

43
dao/src/main/java/org/thingsboard/server/dao/notification/cache/NotificationRuleCacheKey.java

@ -1,43 +0,0 @@
/**
* Copyright © 2016-2023 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.dao.notification.cache;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
import java.io.Serializable;
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class NotificationRuleCacheKey implements Serializable {
private static final long serialVersionUID = 5987113265482170L;
private TenantId tenantId;
private NotificationRuleTriggerType triggerType;
@Override
public String toString() {
return tenantId + "_" + triggerType;
}
}

32
dao/src/main/java/org/thingsboard/server/dao/notification/cache/NotificationRuleCaffeineCache.java

@ -1,32 +0,0 @@
/**
* Copyright © 2016-2023 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.dao.notification.cache;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cache.CacheManager;
import org.springframework.stereotype.Service;
import org.thingsboard.server.cache.CaffeineTbTransactionalCache;
import org.thingsboard.server.common.data.CacheConstants;
@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true)
@Service
public class NotificationRuleCaffeineCache extends CaffeineTbTransactionalCache<NotificationRuleCacheKey, NotificationRuleCacheValue> {
public NotificationRuleCaffeineCache(CacheManager cacheManager) {
super(cacheManager, CacheConstants.NOTIFICATION_RULES_CACHE);
}
}

35
dao/src/main/java/org/thingsboard/server/dao/notification/cache/NotificationRuleRedisCache.java

@ -1,35 +0,0 @@
/**
* Copyright © 2016-2023 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.dao.notification.cache;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.stereotype.Service;
import org.thingsboard.server.cache.CacheSpecsMap;
import org.thingsboard.server.cache.RedisTbTransactionalCache;
import org.thingsboard.server.cache.TBRedisCacheConfiguration;
import org.thingsboard.server.cache.TbFSTRedisSerializer;
import org.thingsboard.server.common.data.CacheConstants;
@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis")
@Service
public class NotificationRuleRedisCache extends RedisTbTransactionalCache<NotificationRuleCacheKey, NotificationRuleCacheValue> {
public NotificationRuleRedisCache(CacheSpecsMap cacheSpecsMap, RedisConnectionFactory connectionFactory, TBRedisCacheConfiguration configuration) {
super(CacheConstants.NOTIFICATION_RULES_CACHE, cacheSpecsMap, connectionFactory, configuration, new TbFSTRedisSerializer<>());
}
}
Loading…
Cancel
Save