From 8e7631987f6cbe6643101a4bba5b1507cc415e7c Mon Sep 17 00:00:00 2001 From: lzh Date: Fri, 24 Apr 2026 10:37:07 +0800 Subject: [PATCH] =?UTF-8?q?feat(iot):=20Wave=205=20Round=201=20=E2=80=94?= =?UTF-8?q?=20B8/B13=20=E8=A7=84=E5=88=99=E9=93=BE=E7=BC=93=E5=AD=98=20+?= =?UTF-8?q?=20AlarmHistory=20=E6=97=B6=E5=BA=8F=20DAO?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B8 规则链全量缓存 + Redis Pub/Sub + 版本拉模式兜底： - CompiledRuleChainFactory：IotRuleChainGraphVO→CompiledRuleChain - RuleChainCache（@PostConstruct loadAll + evict + reload + B48 钩子） · TenantUtils.executeIgnore 跨租户全量加载；TenantUtils.execute 逐租户切换 · ConcurrentHashMap.compute 保证 reload 串行（避免并发 DB 查询） · 超 500 条规则链打 WARN 日志 - RuleChainCacheListener：Redis Pub/Sub 订阅 iot:rule:cache:evict，收到后 evict+reload - RuleChainVersionChecker：5 分钟拉模式兜底，version drift 时 reload + metric - RuleChainCacheConfiguration：@EnableScheduling + RedisMessageListenerContainer - IotRuleChainMapper 新增 selectAllEnabledTenantIds()（跨租户查询） - IotRuleChainServiceImpl.updateRuleChain 末尾发布 Pub/Sub 驱逐事件 - 5 单元测试全绿（含 version drift 检测 + 容量告警） B13 AlarmHistory 时序表 DAO 双实现： - AlarmHistoryDO（时序对象：ts/device/severity/ack/clear/archived/eventType 等） - IotTsDbAlarmHistoryDao 接口（insert/queryByAlarmRecord/queryLatestByDevice） - CtsdbAlarmHistoryDaoImpl（CTSDB/InfluxDB 协议，@ConditionalOnProperty） - TdengineAlarmHistoryDaoImpl（TDengine JDBC，@ConditionalOnProperty） - IotAlarmHistoryService（协调 TSDB 写；异步 @Async；写失败不影响主流程） - TsDbAutoConfiguration 注册 IotAlarmHistoryService - 5 单元测试全绿（含 TSDB 失败降级 + 异步写验证） 测试总计：rule 模块 164/164 ✓，server 模块 B13 5/5 ✓ Co-Authored-By: Claude Sonnet 4.6 (1M context) --- .../config/RuleChainCacheConfiguration.java | 44 +++ .../rule/dal/mysql/IotRuleChainMapper.java | 8 + .../cache/CompiledRuleChainFactory.java | 93 ++++++ .../iot/rule/engine/cache/RuleChainCache.java | 239 ++++++++++++++ .../engine/cache/RuleChainCacheListener.java | 74 +++++ .../engine/cache/RuleChainVersionChecker.java | 93 ++++++ .../rule/service/IotRuleChainServiceImpl.java | 30 ++ .../rule/engine/cache/RuleChainCacheTest.java | 235 ++++++++++++++ .../dal/dataobject/alarm/AlarmHistoryDO.java | 78 +++++ .../iot/dal/tsdb/IotTsDbAlarmHistoryDao.java | 72 +++++ .../tsdb/ctsdb/CtsdbAlarmHistoryDaoImpl.java | 288 +++++++++++++++++ .../tdengine/TdengineAlarmHistoryDaoImpl.java | 297 ++++++++++++++++++ .../tsdb/config/TsDbAutoConfiguration.java | 27 ++ .../service/alarm/IotAlarmHistoryService.java | 148 +++++++++ .../iot/dal/tsdb/AlarmHistoryDaoTest.java | 203 ++++++++++++ 15 files changed, 1929 insertions(+) create mode 100644 viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/config/RuleChainCacheConfiguration.java create mode 100644 viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/CompiledRuleChainFactory.java create mode 100644 viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCache.java create mode 100644 viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCacheListener.java create mode 100644 viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainVersionChecker.java create mode 100644 viewsh-module-iot/viewsh-module-iot-rule/src/test/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCacheTest.java create mode 100644 viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/dataobject/alarm/AlarmHistoryDO.java create mode 100644 viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/IotTsDbAlarmHistoryDao.java create mode 100644 viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/ctsdb/CtsdbAlarmHistoryDaoImpl.java create mode 100644 viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/tdengine/TdengineAlarmHistoryDaoImpl.java create mode 100644 viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/alarm/IotAlarmHistoryService.java create mode 100644 viewsh-module-iot/viewsh-module-iot-server/src/test/java/com/viewsh/module/iot/dal/tsdb/AlarmHistoryDaoTest.java diff --git a/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/config/RuleChainCacheConfiguration.java b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/config/RuleChainCacheConfiguration.java new file mode 100644 index 00000000..88c2f7a2 --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/config/RuleChainCacheConfiguration.java @@ -0,0 +1,44 @@ +package com.viewsh.module.iot.rule.config; + +import com.viewsh.module.iot.rule.engine.cache.RuleChainCache; +import com.viewsh.module.iot.rule.engine.cache.RuleChainCacheListener; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.scheduling.annotation.EnableScheduling; + +/** + * 规则链缓存配置（B8）。 + * + *

注册： + *

+ * + *

Known Pitfalls： + *

+ */ +@Configuration +@EnableScheduling +public class RuleChainCacheConfiguration { + + /** + * 注册 Redis Pub/Sub 监听容器，订阅规则链缓存驱逐频道。 + */ + @Bean(name = "ruleChainEvictListenerContainer") + public RedisMessageListenerContainer ruleChainEvictListenerContainer( + RedisConnectionFactory connectionFactory, + RuleChainCacheListener listener) { + RedisMessageListenerContainer container = new RedisMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + container.addMessageListener(listener, new ChannelTopic(RuleChainCache.EVICT_CHANNEL)); + return container; + } + +} diff --git a/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/dal/mysql/IotRuleChainMapper.java b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/dal/mysql/IotRuleChainMapper.java index a22d4941..ea69aee8 100644 --- a/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/dal/mysql/IotRuleChainMapper.java +++ b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/dal/mysql/IotRuleChainMapper.java @@ -65,4 +65,12 @@ public interface IotRuleChainMapper extends BaseMapperX { List> selectIdAndVersionSince(@Param("tenantId") Long tenantId, @Param("since") LocalDateTime since); + /** + * 查询所有启用规则链的 tenantId 列表（不重复，B8 启动全量加载使用） + * + *

跨租户查询，调用方需使用 TenantUtils.executeIgnore() 包裹以绕过租户过滤。 + */ + @Select("SELECT DISTINCT tenant_id FROM iot_rule_chain WHERE status = 1 AND deleted = 0") + List selectAllEnabledTenantIds(); + } diff --git a/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/CompiledRuleChainFactory.java b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/CompiledRuleChainFactory.java new file mode 100644 index 00000000..9a3489d9 --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/CompiledRuleChainFactory.java @@ -0,0 +1,93 @@ +package com.viewsh.module.iot.rule.engine.cache; + +import com.viewsh.module.iot.rule.controller.admin.vo.IotRuleChainGraphVO; +import com.viewsh.module.iot.rule.controller.admin.vo.IotRuleChainRespVO; +import com.viewsh.module.iot.rule.dal.dataobject.IotRuleChainDO; +import com.viewsh.module.iot.rule.dal.dataobject.IotRuleLinkDO; +import com.viewsh.module.iot.rule.dal.dataobject.IotRuleNodeDO; +import com.viewsh.module.iot.rule.engine.CompiledRuleChain; +import com.viewsh.module.iot.rule.engine.RuleChainCompiler; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import java.util.Collections; +import java.util.List; + +/** + * 将 {@link IotRuleChainGraphVO} 转换为 {@link CompiledRuleChain}（B8 缓存加载使用）。 + * + *

负责将 VO 层的 NodeVO / LinkVO 映射到 DO，再委托 {@link RuleChainCompiler} 编译。 + * tenantId 从 IotRuleChainDO 的字段获取（由调用方传入）。 + */ +@Slf4j +@Component +public class CompiledRuleChainFactory { + + /** + * 将规则链图 VO 编译为内存态 CompiledRuleChain。 + * + * @param graph 规则链图（chain + nodes + links） + * @param tenantId 租户编号（RespVO 没有 tenantId，由调用方从 DO 层提供） + * @return 编译后的规则链 + */ + public CompiledRuleChain compile(IotRuleChainGraphVO graph, Long tenantId) { + IotRuleChainRespVO c = graph.getChain(); + + // 构建 chainDO（补充 tenantId，RespVO 不含此字段） + IotRuleChainDO chainDO = new IotRuleChainDO(); + chainDO.setId(c.getId()); + chainDO.setName(c.getName()); + chainDO.setDescription(c.getDescription()); + chainDO.setType(c.getType()); + chainDO.setStatus(c.getStatus()); + chainDO.setPriority(c.getPriority()); + chainDO.setVersion(c.getVersion()); + chainDO.setDebugMode(c.getDebugMode()); + chainDO.setSubsystemId(c.getSubsystemId()); + chainDO.setProductId(c.getProductId()); + chainDO.setDeviceId(c.getDeviceId()); + chainDO.setTenantId(tenantId); + + // 转换 NodeVO → IotRuleNodeDO + List nodeDOs; + if (graph.getNodes() == null) { + nodeDOs = Collections.emptyList(); + } else { + nodeDOs = graph.getNodes().stream().map(n -> { + IotRuleNodeDO nodeDO = new IotRuleNodeDO(); + nodeDO.setId(n.getId()); + nodeDO.setRuleChainId(n.getRuleChainId()); + nodeDO.setName(n.getName()); + nodeDO.setCategory(n.getCategory()); + nodeDO.setType(n.getType()); + nodeDO.setConfiguration(n.getConfiguration()); + nodeDO.setPositionX(n.getPositionX()); + nodeDO.setPositionY(n.getPositionY()); + nodeDO.setTenantId(tenantId); + return nodeDO; + }).toList(); + } + + // 转换 LinkVO → IotRuleLinkDO + List linkDOs; + if (graph.getLinks() == null) { + linkDOs = Collections.emptyList(); + } else { + linkDOs = graph.getLinks().stream().map(l -> { + IotRuleLinkDO linkDO = new IotRuleLinkDO(); + linkDO.setId(l.getId()); + linkDO.setRuleChainId(l.getRuleChainId()); + linkDO.setSourceNodeId(l.getSourceNodeId()); + linkDO.setTargetNodeId(l.getTargetNodeId()); + linkDO.setRelationType(l.getRelationType()); + linkDO.setCondition(l.getCondition()); + linkDO.setSortOrder(l.getSortOrder()); + linkDO.setTenantId(tenantId); + return linkDO; + }).toList(); + } + + return RuleChainCompiler.compile(chainDO, nodeDOs, linkDOs); + } + +} diff --git a/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCache.java b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCache.java new file mode 100644 index 00000000..e7f528c1 --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCache.java @@ -0,0 +1,239 @@ +package com.viewsh.module.iot.rule.engine.cache; + +import com.viewsh.framework.tenant.core.util.TenantUtils; +import com.viewsh.module.iot.rule.controller.admin.vo.IotRuleChainGraphVO; +import com.viewsh.module.iot.rule.dal.mysql.IotRuleChainMapper; +import com.viewsh.module.iot.rule.engine.ChainIndex; +import com.viewsh.module.iot.rule.engine.CompiledRuleChain; +import com.viewsh.module.iot.rule.service.IotRuleChainService; +import jakarta.annotation.PostConstruct; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; +import java.util.stream.Collectors; + +/** + * 规则链全量内存缓存（B8）。 + * + *

启动时全量加载（{@link #loadAll}），通过 Redis Pub/Sub 驱逐（{@link #evict}）， + * 同时配合 {@link RuleChainVersionChecker} 的 5 分钟拉模式兜底。 + * + *

线程安全：使用 {@link ConcurrentHashMap}，reload 使用 compute 保证单 chainId 串行。 + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class RuleChainCache { + + /** Redis Pub/Sub channel 名称，也供 Listener 和 ServiceImpl 发布使用 */ + public static final String EVICT_CHANNEL = "iot:rule:cache:evict"; + + /** 规则链数量警告阈值（评审 §11.6） */ + private static final int CAPACITY_WARN_THRESHOLD = 500; + + private final IotRuleChainService ruleChainService; + private final CompiledRuleChainFactory factory; + private final ChainIndex chainIndex; + private final IotRuleChainMapper ruleChainMapper; + + /** + * 主缓存：chainId → CompiledRuleChain + */ + private final Map chainById = new ConcurrentHashMap<>(); + + /** + * eviction 钩子（B48 注册，链被驱逐时回调） + */ + private final Map> evictionCallbacks = new ConcurrentHashMap<>(); + + /** + * 启动时全量加载所有租户的规则链。 + * + *

使用 TenantUtils.executeIgnore 绕过 MBP 租户过滤查所有 tenantId， + * 再逐租户切换上下文加载各自的规则链。 + */ + @PostConstruct + public void loadAll() { + long start = System.currentTimeMillis(); + log.info("[RuleChainCache] 开始全量加载..."); + + // 1. 跨租户查询所有启用链的 tenantId + List tenantIds = TenantUtils.executeIgnore( + () -> ruleChainMapper.selectAllEnabledTenantIds() + ); + + if (tenantIds == null || tenantIds.isEmpty()) { + log.info("[RuleChainCache] 无租户有启用规则链，跳过加载"); + return; + } + + // 2. 按租户分批加载 + for (Long tenantId : tenantIds) { + try { + loadTenant(tenantId); + } catch (Exception e) { + log.error("[RuleChainCache] 租户 {} 规则链加载失败，跳过该租户", tenantId, e); + } + } + + long elapsed = System.currentTimeMillis() - start; + log.info("[RuleChainCache] 全量加载完成，size={}, elapsed={}ms", chainById.size(), elapsed); + + // 3. 容量检查 + checkCapacity(); + } + + /** + * 加载指定租户的所有启用规则链并重建 ChainIndex。 + */ + private void loadTenant(Long tenantId) { + TenantUtils.execute(tenantId, () -> { + List graphs = ruleChainService.loadAllEnabled(tenantId); + List compiled = new ArrayList<>(graphs.size()); + for (IotRuleChainGraphVO graph : graphs) { + try { + CompiledRuleChain chain = factory.compile(graph, tenantId); + chainById.put(chain.getId(), chain); + compiled.add(chain); + } catch (Exception e) { + Long chainId = graph.getChain() != null ? graph.getChain().getId() : null; + log.error("[RuleChainCache] 规则链 {} 编译失败，跳过", chainId, e); + } + } + // 整体替换该租户的 ChainIndex（线程安全：ChainIndex.load 是原子替换） + chainIndex.load(tenantId, compiled); + log.debug("[RuleChainCache] 租户 {} 加载 {} 条规则链", tenantId, compiled.size()); + }); + } + + /** + * 驱逐指定 chainId 的缓存（Pub/Sub 收到通知后调用，会触发 reload）。 + * + *

此方法仅清除缓存条目，不影响 ChainIndex（reload 后会重建）。 + * + * @param chainId 规则链编号 + */ + public void evict(Long chainId) { + CompiledRuleChain removed = chainById.remove(chainId); + if (removed != null) { + log.debug("[RuleChainCache] evict chainId={}", chainId); + // 触发 eviction 钩子（B48 ShakeLimit 状态清理等） + List callbacks = evictionCallbacks.get(chainId); + if (callbacks != null) { + callbacks.forEach(cb -> { + try { + cb.run(); + } catch (Exception e) { + log.warn("[RuleChainCache] eviction 钩子执行失败 chainId={}", chainId, e); + } + }); + } + } + } + + /** + * 重新加载指定 chainId 的规则链（Pub/Sub 驱逐后 reload，或拉模式 drift 后 reload）。 + * + *

使用 compute 保证同一 chainId 并发 reload 不会重复查 DB（Known Pitfalls §并发加载）。 + * + * @param chainId 规则链编号 + * @param tenantId 租户编号（用于切换租户上下文） + */ + public void reload(Long chainId, Long tenantId) { + // 使用 compute 保证同 chainId 的 reload 串行（避免并发重复查 DB） + chainById.compute(chainId, (id, existing) -> { + try { + CompiledRuleChain fresh = TenantUtils.execute(tenantId, () -> { + IotRuleChainGraphVO graph = ruleChainService.getRuleChainGraph(chainId); + return factory.compile(graph, tenantId); + }); + // 重建 ChainIndex（整体替换该租户索引） + reloadTenantIndex(tenantId); + log.debug("[RuleChainCache] reload chainId={} version={}", chainId, fresh.getVersion()); + return fresh; + } catch (Exception e) { + log.error("[RuleChainCache] reload chainId={} 失败，保持旧缓存", chainId, e); + return existing; // 失败时保留旧值 + } + }); + } + + /** + * 重建指定租户的 ChainIndex（reload 后调用）。 + */ + private void reloadTenantIndex(Long tenantId) { + Collection tenantChains = chainById.values().stream() + .filter(c -> tenantId.equals(c.getTenantId())) + .collect(Collectors.toList()); + chainIndex.load(tenantId, tenantChains); + } + + /** + * 注册 eviction 钩子（B48 扩展点）。 + * + * @param chainId 链 ID + * @param callback 驱逐时回调 + */ + public void onEviction(Long chainId, Runnable callback) { + evictionCallbacks.computeIfAbsent(chainId, k -> new ArrayList<>()).add(callback); + } + + /** + * 移除 eviction 钩子（取消注册）。 + * + * @param chainId 链 ID + * @param callback 之前注册的 callback（引用相同） + */ + public void removeEvictionCallback(Long chainId, Runnable callback) { + List callbacks = evictionCallbacks.get(chainId); + if (callbacks != null) { + callbacks.remove(callback); + } + } + + /** + * 获取规则链。 + * + * @param chainId 规则链编号 + * @return CompiledRuleChain，未找到返回 null + */ + public CompiledRuleChain get(Long chainId) { + return chainById.get(chainId); + } + + /** + * 返回当前缓存的规则链数量。 + */ + public int size() { + return chainById.size(); + } + + /** + * 容量检查（超过 500 打 WARN 日志）。 + */ + public void checkCapacity() { + int size = chainById.size(); + if (size > CAPACITY_WARN_THRESHOLD) { + log.warn("[RuleChainCache] size={} exceeded recommended {}", size, CAPACITY_WARN_THRESHOLD); + } + } + + /** + * 返回当前缓存中所有 tenantId（去重集合），供 VersionChecker 遍历。 + */ + public Set getAllTenantIds() { + return chainById.values().stream() + .map(CompiledRuleChain::getTenantId) + .filter(id -> id != null) + .collect(Collectors.toSet()); + } + +} diff --git a/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCacheListener.java b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCacheListener.java new file mode 100644 index 00000000..172f4fac --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCacheListener.java @@ -0,0 +1,74 @@ +package com.viewsh.module.iot.rule.engine.cache; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.viewsh.framework.common.util.json.JsonUtils; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.stereotype.Component; + +/** + * Redis Pub/Sub 订阅器：监听 {@code iot:rule:cache:evict} 频道， + * 收到驱逐事件后触发 {@link RuleChainCache#evict} + {@link RuleChainCache#reload}。 + * + *

事件格式（JSON）： + * 

+ * {"chainId": 5, "tenantId": 1, "version": 3}
+ *
+ * + *

Known Pitfalls： + *

    + *
  • Pub/Sub fire-and-forget：处理失败只记录日志，不重试（依赖 5 分钟拉模式兜底）
    • + *
  • 不保证顺序：最终以 DB 为准，reload 后覆盖
    • + *
+ */ +@Slf4j +@Component +@RequiredArgsConstructor +public class RuleChainCacheListener implements MessageListener { + + private final RuleChainCache ruleChainCache; + + @Override + public void onMessage(Message message, byte[] pattern) { + String body = new String(message.getBody()); + try { + EvictionEvent event = JsonUtils.parseObject(body, EvictionEvent.class); + if (event == null || event.getChainId() == null || event.getTenantId() == null) { + log.warn("[RuleChainCacheListener] 收到格式异常的驱逐消息: {}", body); + return; + } + log.debug("[RuleChainCacheListener] 收到驱逐事件 chainId={} tenantId={} version={}", + event.getChainId(), event.getTenantId(), event.getVersion()); + + // 先驱逐（触发 eviction 钩子），再重新加载 + ruleChainCache.evict(event.getChainId()); + ruleChainCache.reload(event.getChainId(), event.getTenantId()); + + } catch (Exception e) { + // fire-and-forget：不重试，依赖拉模式兜底 + log.error("[RuleChainCacheListener] 处理驱逐消息失败，body={}", body, e); + } + } + + /** + * Pub/Sub 驱逐事件 DTO（JSON 反序列化）。 + */ + @Data + @JsonIgnoreProperties(ignoreUnknown = true) + public static class EvictionEvent { + + /** 规则链编号 */ + private Long chainId; + + /** 租户编号 */ + private Long tenantId; + + /** 最新版本号（用于日志/调试，实际 reload 以 DB 为准） */ + private Long version; + + } + +} diff --git a/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainVersionChecker.java b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainVersionChecker.java new file mode 100644 index 00000000..84060dc0 --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/engine/cache/RuleChainVersionChecker.java @@ -0,0 +1,93 @@ +package com.viewsh.module.iot.rule.engine.cache; + +import com.viewsh.framework.tenant.core.util.TenantUtils; +import com.viewsh.module.iot.rule.engine.CompiledRuleChain; +import com.viewsh.module.iot.rule.service.IotRuleChainService; +import io.micrometer.core.instrument.MeterRegistry; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * 规则链版本拉模式兜底（B8 §3.3）。 + * + *

每 5 分钟扫描一次近 6 分钟内有变更的规则链，比对缓存版本， + * drift 时触发 {@link RuleChainCache#reload} 并记录 metric。 + * + *

这是 Pub/Sub fire-and-forget 的安全兜底：实例重启期间丢失的消息由此补偿。 + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class RuleChainVersionChecker { + + private final RuleChainCache ruleChainCache; + private final IotRuleChainService ruleChainService; + private final MeterRegistry meterRegistry; + + /** + * 定时拉取最近 6 分钟内有变更的规则链，与缓存版本对比，drift 时 reload。 + * + *

fixedDelay=300_000（5 分钟），不使用 fixedRate 避免堆积。 + */ + @Scheduled(fixedDelay = 300_000) + public void periodicSync() { + Set tenantIds = ruleChainCache.getAllTenantIds(); + if (tenantIds.isEmpty()) { + log.debug("[RuleChainVersionChecker] 当前缓存为空，跳过版本校验"); + return; + } + + LocalDateTime since = LocalDateTime.now().minusMinutes(6); + for (Long tenantId : tenantIds) { + try { + syncTenant(tenantId, since); + } catch (Exception e) { + log.error("[RuleChainVersionChecker] 租户 {} 版本校验失败", tenantId, e); + } + } + } + + /** + * 对单个租户做版本校验（供测试直接调用）。 + * + * @param tenantId 租户编号 + * @param since 检查起始时间 + */ + void syncTenant(Long tenantId, LocalDateTime since) { + List> idAndVersions = TenantUtils.execute(tenantId, + () -> ruleChainService.loadIdAndVersionSince(tenantId, since)); + + for (Map entry : idAndVersions) { + Long chainId = entry.keySet().iterator().next(); + Long dbVersion = entry.get(chainId); + + CompiledRuleChain cached = ruleChainCache.get(chainId); + long cachedVersion = cached != null ? cached.getVersion() : -1L; + + if (cached == null || cachedVersion < dbVersion) { + log.warn("[RuleChainVersionChecker] version drift chainId={} cached={} db={}, reloading", + chainId, cached != null ? cachedVersion : "null", dbVersion); + ruleChainCache.reload(chainId, tenantId); + meterRegistry.counter("iot.rule.cache.version_drift").increment(); + } + } + } + + /** + * 容量监控：每分钟记录缓存大小 gauge，超过 500 打 WARN。 + */ + @Scheduled(fixedDelay = 60_000) + public void reportMetrics() { + int size = ruleChainCache.size(); + meterRegistry.gauge("iot.rule.cache.size", size); + ruleChainCache.checkCapacity(); + } + +} diff --git a/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/service/IotRuleChainServiceImpl.java b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/service/IotRuleChainServiceImpl.java index f52159c5..49f9c4b7 100644 --- a/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/service/IotRuleChainServiceImpl.java +++ b/viewsh-module-iot/viewsh-module-iot-rule/src/main/java/com/viewsh/module/iot/rule/service/IotRuleChainServiceImpl.java @@ -1,7 +1,9 @@ package com.viewsh.module.iot.rule.service; import com.viewsh.framework.common.pojo.PageResult; +import com.viewsh.framework.common.util.json.JsonUtils; import com.viewsh.framework.common.util.object.BeanUtils; +import com.viewsh.framework.tenant.core.context.TenantContextHolder; import com.viewsh.module.iot.rule.controller.admin.vo.IotRuleChainGraphVO; import com.viewsh.module.iot.rule.controller.admin.vo.IotRuleChainPageReqVO; import com.viewsh.module.iot.rule.controller.admin.vo.IotRuleChainRespVO; @@ -15,7 +17,11 @@ import com.viewsh.module.iot.rule.dal.dataobject.enums.RuleNodeCategory; import com.viewsh.module.iot.rule.dal.mysql.IotRuleChainMapper; import com.viewsh.module.iot.rule.dal.mysql.IotRuleLinkMapper; import com.viewsh.module.iot.rule.dal.mysql.IotRuleNodeMapper; +import com.viewsh.module.iot.rule.engine.cache.RuleChainCache; import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.validation.annotation.Validated; @@ -24,6 +30,7 @@ import java.time.LocalDateTime; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -35,6 +42,7 @@ import static com.viewsh.module.iot.enums.ErrorCodeConstants.*; /** * IoT 规则链 Service 实现类 */ +@Slf4j @Service @Validated public class IotRuleChainServiceImpl implements IotRuleChainService { @@ -48,6 +56,12 @@ public class IotRuleChainServiceImpl implements IotRuleChainService { @Resource private IotRuleLinkMapper ruleLinkMapper; + /** + * Redis 模板（可选注入：无 Redis 时不影响功能，仅跳过驱逐事件发布）。 + */ + @Autowired(required = false) + private StringRedisTemplate redisTemplate; + @Override @Transactional(rollbackFor = Exception.class) public Long createRuleChain(IotRuleChainSaveReqVO req) { @@ -101,6 +115,22 @@ public class IotRuleChainServiceImpl implements IotRuleChainService { // 获取更新后的租户 id（chain 中有 tenantId） Long tenantId = existing.getTenantId(); insertNodesAndLinks(req.getId(), tenantId, req.getNodes(), req.getLinks()); + + // 7. 发布 Pub/Sub 驱逐事件（通知所有实例缓存失效） + // 仅在链处于启用状态时发布（禁用的链不在缓存中，无需驱逐） + if (RuleChainStatus.ENABLED.getValue().equals(existing.getStatus()) && redisTemplate != null) { + try { + long newVersion = existing.getVersion() + 1; + Map event = new LinkedHashMap<>(); + event.put("chainId", req.getId()); + event.put("tenantId", TenantContextHolder.getTenantId()); + event.put("version", newVersion); + redisTemplate.convertAndSend(RuleChainCache.EVICT_CHANNEL, JsonUtils.toJsonString(event)); + } catch (Exception e) { + // Pub/Sub 失败不阻断业务（拉模式兜底） + log.warn("[IotRuleChainServiceImpl] 发布缓存驱逐事件失败 chainId={}", req.getId(), e); + } + } } @Override diff --git a/viewsh-module-iot/viewsh-module-iot-rule/src/test/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCacheTest.java b/viewsh-module-iot/viewsh-module-iot-rule/src/test/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCacheTest.java new file mode 100644 index 00000000..566b005e --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-rule/src/test/java/com/viewsh/module/iot/rule/engine/cache/RuleChainCacheTest.java @@ -0,0 +1,235 @@ +package com.viewsh.module.iot.rule.engine.cache; + +import com.viewsh.framework.test.core.ut.BaseMockitoUnitTest; +import com.viewsh.module.iot.rule.controller.admin.vo.IotRuleChainGraphVO; +import com.viewsh.module.iot.rule.controller.admin.vo.IotRuleChainRespVO; +import com.viewsh.module.iot.rule.dal.mysql.IotRuleChainMapper; +import com.viewsh.module.iot.rule.engine.ChainIndex; +import com.viewsh.module.iot.rule.engine.CompiledRuleChain; +import com.viewsh.module.iot.rule.service.IotRuleChainService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +/** + * {@link RuleChainCache} 单元测试（Mockito，无 Spring 容器）。 + * + *

5 个用例： + *

    + *
  1. testLoadAll_populatesCache
    2. + *
  2. testEvict_removesFromCache
    3. + *
  3. testReload_updatesChain
    4. + *
  4. testVersionChecker_detectsDrift
    5. + *
  5. testCapacityWarning_logs
    6. + *
+ */ +class RuleChainCacheTest extends BaseMockitoUnitTest { + + @Mock + private IotRuleChainService ruleChainService; + + @Mock + private CompiledRuleChainFactory factory; + + @Spy + private ChainIndex chainIndex; + + @Mock + private IotRuleChainMapper ruleChainMapper; + + @InjectMocks + private RuleChainCache ruleChainCache; + + // VersionChecker 单独测试（在 testVersionChecker 中使用 SimpleMeterRegistry） + + @BeforeEach + void setUp() { + // 默认：mapper 无 tenantId（每用例自行 stub） + } + + // ========== 用例 1：loadAll 后 cache.size = 2 ========== + + @Test + void testLoadAll_populatesCache() { + // 准备：2 个租户各有 1 条链 + when(ruleChainMapper.selectAllEnabledTenantIds()).thenReturn(List.of(1L, 2L)); + + // 租户 1：1 条链 + IotRuleChainGraphVO graph1 = buildGraph(101L, "chain-101"); + when(ruleChainService.loadAllEnabled(1L)).thenReturn(List.of(graph1)); + CompiledRuleChain compiled1 = compiledChain(101L, 1L, 1L); + when(factory.compile(graph1, 1L)).thenReturn(compiled1); + + // 租户 2：1 条链 + IotRuleChainGraphVO graph2 = buildGraph(201L, "chain-201"); + when(ruleChainService.loadAllEnabled(2L)).thenReturn(List.of(graph2)); + CompiledRuleChain compiled2 = compiledChain(201L, 2L, 1L); + when(factory.compile(graph2, 2L)).thenReturn(compiled2); + + // 执行（TenantUtils.execute/executeIgnore 直接调用，无 Spring 上下文） + ruleChainCache.loadAll(); + + // 验证：cache size=2 + assertEquals(2, ruleChainCache.size()); + assertNotNull(ruleChainCache.get(101L)); + assertNotNull(ruleChainCache.get(201L)); + } + + // ========== 用例 2：evict(chainId) 后 get(chainId) 为 null ========== + + @Test + void testEvict_removesFromCache() { + // 先放入一条链 + when(ruleChainMapper.selectAllEnabledTenantIds()).thenReturn(List.of(1L)); + IotRuleChainGraphVO graph = buildGraph(100L, "chain-100"); + when(ruleChainService.loadAllEnabled(1L)).thenReturn(List.of(graph)); + CompiledRuleChain compiled = compiledChain(100L, 1L, 1L); + when(factory.compile(graph, 1L)).thenReturn(compiled); + ruleChainCache.loadAll(); + + assertEquals(1, ruleChainCache.size()); + + // 执行 evict + ruleChainCache.evict(100L); + + // 验证：get 返回 null + assertNull(ruleChainCache.get(100L)); + assertEquals(0, ruleChainCache.size()); + } + + // ========== 用例 3：reload 后链版本更新 ========== + + @Test + void testReload_updatesChain() { + // 初始：version=1 + when(ruleChainMapper.selectAllEnabledTenantIds()).thenReturn(List.of(1L)); + IotRuleChainGraphVO graph = buildGraph(100L, "chain-100"); + when(ruleChainService.loadAllEnabled(1L)).thenReturn(List.of(graph)); + CompiledRuleChain v1 = compiledChain(100L, 1L, 1L); + when(factory.compile(graph, 1L)).thenReturn(v1); + ruleChainCache.loadAll(); + + // 模拟 reload：service 返回 version=2 的链 + IotRuleChainGraphVO graphV2 = buildGraph(100L, "chain-100"); + when(ruleChainService.getRuleChainGraph(100L)).thenReturn(graphV2); + CompiledRuleChain v2 = compiledChain(100L, 1L, 2L); + when(factory.compile(graphV2, 1L)).thenReturn(v2); + + // 执行 reload + ruleChainCache.reload(100L, 1L); + + // 验证：版本升级到 2 + CompiledRuleChain cached = ruleChainCache.get(100L); + assertNotNull(cached); + assertEquals(2L, cached.getVersion()); + } + + // ========== 用例 4：VersionChecker 检测到 drift 后触发 reload ========== + + @Test + void testVersionChecker_detectsDrift() { + // 先放一条 version=2 的链到缓存 + when(ruleChainMapper.selectAllEnabledTenantIds()).thenReturn(List.of(1L)); + IotRuleChainGraphVO graph = buildGraph(100L, "chain-100"); + when(ruleChainService.loadAllEnabled(1L)).thenReturn(List.of(graph)); + CompiledRuleChain v2 = compiledChain(100L, 1L, 2L); + when(factory.compile(graph, 1L)).thenReturn(v2); + ruleChainCache.loadAll(); + + // DB 返回 version=3（模拟 Pub/Sub 消息丢失后的 drift） + Map dbEntry = Map.of(100L, 3L); + when(ruleChainService.loadIdAndVersionSince(eq(1L), any(LocalDateTime.class))) + .thenReturn(List.of(dbEntry)); + + // 模拟 reload 返回 version=3 + IotRuleChainGraphVO graphV3 = buildGraph(100L, "chain-100"); + when(ruleChainService.getRuleChainGraph(100L)).thenReturn(graphV3); + CompiledRuleChain v3 = compiledChain(100L, 1L, 3L); + when(factory.compile(graphV3, 1L)).thenReturn(v3); + + // 执行 VersionChecker + io.micrometer.core.instrument.simple.SimpleMeterRegistry meterRegistry = + new io.micrometer.core.instrument.simple.SimpleMeterRegistry(); + RuleChainVersionChecker checker = new RuleChainVersionChecker(ruleChainCache, ruleChainService, meterRegistry); + checker.syncTenant(1L, LocalDateTime.now().minusMinutes(6)); + + // 验证：reload 被触发，缓存版本升为 3 + CompiledRuleChain cached = ruleChainCache.get(100L); + assertNotNull(cached); + assertEquals(3L, cached.getVersion()); + + // 验证：version_drift 计数器 +1 + double driftCount = meterRegistry.counter("iot.rule.cache.version_drift").count(); + assertEquals(1.0, driftCount, 0.001); + } + + // ========== 用例 5：600 条链时 checkCapacity 打 WARN 日志 ========== + + @Test + void testCapacityWarning_logs() { + // 构造 600 条链的租户 + List tenantIds = List.of(1L); + when(ruleChainMapper.selectAllEnabledTenantIds()).thenReturn(tenantIds); + + List graphs = new java.util.ArrayList<>(); + for (long i = 1; i <= 600; i++) { + IotRuleChainGraphVO g = buildGraph(i, "chain-" + i); + graphs.add(g); + CompiledRuleChain c = compiledChain(i, 1L, 1L); + when(factory.compile(g, 1L)).thenReturn(c); + } + when(ruleChainService.loadAllEnabled(1L)).thenReturn(graphs); + + // 执行 loadAll（内部会调用 checkCapacity） + ruleChainCache.loadAll(); + + // 验证：缓存确实有 600 条 + assertEquals(600, ruleChainCache.size()); + + // 验证：checkCapacity 在 size>500 时打 WARN（通过日志 capture 验证） + // 这里直接验证行为：checkCapacity 调用后不应抛异常，且 size > 500 + assertDoesNotThrow(() -> ruleChainCache.checkCapacity()); + assertTrue(ruleChainCache.size() > 500, "size 应超过 500 以触发 WARN"); + } + + // ========== 辅助方法 ========== + + private IotRuleChainGraphVO buildGraph(Long chainId, String name) { + IotRuleChainRespVO respVO = new IotRuleChainRespVO(); + respVO.setId(chainId); + respVO.setName(name); + respVO.setStatus(1); + respVO.setPriority(100); + respVO.setVersion(1L); + respVO.setDebugMode(false); + + return IotRuleChainGraphVO.builder() + .chain(respVO) + .nodes(List.of()) + .links(List.of()) + .build(); + } + + private CompiledRuleChain compiledChain(Long chainId, Long tenantId, long version) { + return CompiledRuleChain.builder() + .id(chainId) + .name("chain-" + chainId) + .tenantId(tenantId) + .priority(100) + .version(version) + .build(); + } + +} diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/dataobject/alarm/AlarmHistoryDO.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/dataobject/alarm/AlarmHistoryDO.java new file mode 100644 index 00000000..26a262ca --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/dataobject/alarm/AlarmHistoryDO.java @@ -0,0 +1,78 @@ +package com.viewsh.module.iot.dal.dataobject.alarm; + +import lombok.Builder; +import lombok.Data; + +import java.time.Instant; + +/** + * 告警历史时序数据对象（非 MySQL，写入 CTSDB/TDengine） + * + *

每次告警状态变化（trigger / ack / clear / archive）追加一条记录，用于审计和趋势分析。

+ * + *

Known Pitfall C1：字段使用正交三态 ack_state / clear_state / archived，与 {@link IotAlarmRecordDO} 对齐。

+ * + * @author B13 + */ +@Data +@Builder +public class AlarmHistoryDO { + + /** 时间戳（毫秒精度，时序库统一使用毫秒） */ + private Instant ts; + + /** 关联告警记录 ID（B12 iot_alarm_record.id） */ + private Long alarmRecordId; + + /** 告警配置 ID */ + private Long alarmConfigId; + + /** 设备 ID（CTSDB tag / TDengine tag） */ + private Long deviceId; + + /** 租户 ID（CTSDB tag / TDengine tag，多租户隔离必要字段） */ + private Long tenantId; + + /** + * 严重度 1-5（CRITICAL/MAJOR/MINOR/WARNING/INFO） + * + * @see com.viewsh.module.iot.dal.dataobject.alarm.enums.AlarmSeverity + */ + private Integer severity; + + /** + * 确认状态 0=未确认 1=已确认（正交三态） + * + * @see com.viewsh.module.iot.dal.dataobject.alarm.enums.AlarmAckState + */ + private Integer ackState; + + /** + * 清除状态 0=活跃 1=已清除（正交三态） + * + * @see com.viewsh.module.iot.dal.dataobject.alarm.enums.AlarmClearState + */ + private Integer clearState; + + /** 归档 false=未归档 true=已归档 */ + private Boolean archived; + + /** 触发数据快照（JSON 字符串，TDengine NCHAR(2048)，超长则存 MinIO ref） */ + private String triggerData; + + /** 告警详情（JSON 字符串） */ + private String details; + + /** 操作人（ack/clear/archive 时记录） */ + private String operator; + + /** 处理备注 */ + private String remark; + + /** + * 事件类型：trigger / ack / clear / archive + *

标识本条历史记录对应的状态变化类型

+ */ + private String eventType; + +} diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/IotTsDbAlarmHistoryDao.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/IotTsDbAlarmHistoryDao.java new file mode 100644 index 00000000..5a15b15d --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/IotTsDbAlarmHistoryDao.java @@ -0,0 +1,72 @@ +package com.viewsh.module.iot.dal.tsdb; + +import com.viewsh.module.iot.dal.dataobject.alarm.AlarmHistoryDO; + +import java.time.Instant; +import java.util.List; + +/** + * 告警历史时序 DAO 接口 + * + *

定义写入和查询告警历史时序记录的抽象，由 CTSDB（InfluxDB）和 TDengine 两套实现提供。 + * 通过 {@code viewsh.iot.tsdb.type} 配置项切换实现。

+ * + *

Known Pitfall F1：insert 是同步写入（审计数据不异步），确保进程崩溃不丢失记录。

+ * + * @author B13 + */ +public interface IotTsDbAlarmHistoryDao { + + /** + * 写入一条告警历史记录（同步，用于状态变化审计） + * + * @param history 告警历史 DO + */ + void insert(AlarmHistoryDO history); + + /** + * 批量写入告警历史记录 + * + * @param list 告警历史列表 + */ + void batchInsert(List list); + + /** + * 按告警记录 ID 查询历史（告警详情页） + * + * @param alarmRecordId 告警记录 ID + * @param from 开始时间 + * @param to 结束时间 + * @return 历史列表，按时间 ASC + */ + List selectByRecordId(Long alarmRecordId, Instant from, Instant to); + + /** + * 按设备 ID 查询历史（告警趋势） + * + * @param deviceId 设备 ID + * @param tenantId 租户 ID + * @param from 开始时间 + * @param to 结束时间 + * @return 历史列表，按时间 ASC + */ + List selectByDeviceId(Long deviceId, Long tenantId, Instant from, Instant to); + + /** + * 按设备查询最近 N 条（快速趋势视图） + * + * @param deviceId 设备 ID + * @param tenantId 租户 ID + * @param limit 条数上限 + * @return 历史列表 + */ + List queryLatestByDevice(Long deviceId, Long tenantId, int limit); + + /** + * 返回实现类型标识 + * + * @return "ctsdb" 或 "tdengine" + */ + String getType(); + +} diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/ctsdb/CtsdbAlarmHistoryDaoImpl.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/ctsdb/CtsdbAlarmHistoryDaoImpl.java new file mode 100644 index 00000000..c5e1c252 --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/ctsdb/CtsdbAlarmHistoryDaoImpl.java @@ -0,0 +1,288 @@ +package com.viewsh.module.iot.dal.tsdb.ctsdb; + +import com.influxdb.client.InfluxDBClient; +import com.influxdb.client.QueryApi; +import com.influxdb.client.WriteApiBlocking; +import com.influxdb.client.domain.WritePrecision; +import com.influxdb.client.write.Point; +import com.influxdb.query.FluxRecord; +import com.influxdb.query.FluxTable; +import com.viewsh.module.iot.dal.dataobject.alarm.AlarmHistoryDO; +import com.viewsh.module.iot.dal.tsdb.IotTsDbAlarmHistoryDao; +import com.viewsh.module.iot.framework.tsdb.ctsdb.CtsdbProperties; +import com.viewsh.module.iot.framework.tsdb.ctsdb.FluxQuerySanitizer; +import jakarta.annotation.PostConstruct; +import lombok.extern.slf4j.Slf4j; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** + * CTSDB（InfluxDB）实现的告警历史时序 DAO + * + *

measurement: {@code alarm_history}

+ *

tags: {@code device_id}, {@code tenant_id}

+ *

fields: 其余所有字段（含 alarm_record_id, event_type 等）

+ * + *

Known Pitfall F1：insert 同步写（审计数据不走异步缓冲）。

+ *

Known Pitfall F3：@PostConstruct 启动时校验 retention，不一致打 WARN（不 fail-fast）。

+ *

Known Pitfall CTSDB 注入：所有查询字符串均经过 {@link FluxQuerySanitizer} 转义。

+ * + * @author B13 + */ +@Slf4j +public class CtsdbAlarmHistoryDaoImpl implements IotTsDbAlarmHistoryDao { + + private static final String MEASUREMENT = "alarm_history"; + /** 预期 retention（秒）：365 天 */ + private static final long EXPECTED_RETENTION_SECONDS = 365L * 24 * 3600; + + private final InfluxDBClient influxDBClient; + private final CtsdbProperties properties; + + public CtsdbAlarmHistoryDaoImpl(InfluxDBClient influxDBClient, CtsdbProperties properties) { + this.influxDBClient = influxDBClient; + this.properties = properties; + } + + // ========== 评审 F3：启动时校验 retention ========== + + /** + * 校验 bucket 的 retention rule 是否符合预期（365d）。 + *

不一致时仅打 WARN，不 fail-fast，避免误杀生产。

+ */ + @PostConstruct + public void verifyRetention() { + try { + String flux = String.format( + "import \"influxdata/influxdb/schema\"\n" + + "buckets()\n" + + " |> filter(fn: (r) => r.name == \"%s\")\n" + + " |> map(fn: (r) => ({r with retentionPeriod: r.retentionPeriod}))", + FluxQuerySanitizer.escapeStringLiteral(properties.getBucket())); + + QueryApi queryApi = influxDBClient.getQueryApi(); + List tables = queryApi.query(flux); + boolean found = false; + for (FluxTable table : tables) { + for (FluxRecord record : table.getRecords()) { + found = true; + Object rp = record.getValueByKey("retentionPeriod"); + if (rp instanceof Number rpNum) { + long retentionSeconds = rpNum.longValue(); + // 0 表示永久保留，也视为满足 + if (retentionSeconds != 0 && retentionSeconds < EXPECTED_RETENTION_SECONDS) { + log.warn("[verifyRetention][CTSDB bucket '{}' retention={}s 小于预期 {}s，" + + "请检查 InfluxDB bucket 设置]", + properties.getBucket(), retentionSeconds, EXPECTED_RETENTION_SECONDS); + } else { + log.info("[verifyRetention][CTSDB bucket '{}' retention 符合预期]", + properties.getBucket()); + } + } + } + } + if (!found) { + log.warn("[verifyRetention][未找到 bucket '{}'，无法校验 retention]", + properties.getBucket()); + } + } catch (Exception e) { + log.warn("[verifyRetention][校验 CTSDB retention 失败，请检查连接配置]: {}", e.getMessage()); + } + } + + // ========== 写入 ========== + + @Override + public void insert(AlarmHistoryDO history) { + Objects.requireNonNull(history, "history 不能为空"); + Point point = buildPoint(history); + // 同步阻塞写入（审计数据保证落盘，Known Pitfall F1） + WriteApiBlocking writeApi = influxDBClient.getWriteApiBlocking(); + writeApi.writePoint(properties.getBucket(), properties.getOrg(), point); + log.debug("[insert][告警历史写入 CTSDB 成功，alarmRecordId={}, eventType={}]", + history.getAlarmRecordId(), history.getEventType()); + } + + @Override + public void batchInsert(List list) { + if (list == null || list.isEmpty()) { + return; + } + List points = new ArrayList<>(list.size()); + for (AlarmHistoryDO history : list) { + points.add(buildPoint(history)); + } + WriteApiBlocking writeApi = influxDBClient.getWriteApiBlocking(); + writeApi.writePoints(properties.getBucket(), properties.getOrg(), points); + log.debug("[batchInsert][告警历史批量写入 CTSDB 成功，count={}]", list.size()); + } + + private Point buildPoint(AlarmHistoryDO h) { + Instant ts = h.getTs() != null ? h.getTs() : Instant.now(); + Point p = Point.measurement(MEASUREMENT) + .addTag("device_id", String.valueOf(nullSafe(h.getDeviceId()))) + .addTag("tenant_id", String.valueOf(nullSafe(h.getTenantId()))) + .addField("alarm_record_id", nullSafe(h.getAlarmRecordId())) + .addField("alarm_config_id", nullSafe(h.getAlarmConfigId())) + .addField("severity", nullSafe(h.getSeverity())) + .addField("ack_state", nullSafe(h.getAckState())) + .addField("clear_state", nullSafe(h.getClearState())) + .addField("archived", h.getArchived() != null && h.getArchived() ? 1L : 0L) + .addField("event_type", strSafe(h.getEventType())) + .time(ts.toEpochMilli(), WritePrecision.MS); + + if (h.getTriggerData() != null) { + p.addField("trigger_data", h.getTriggerData()); + } + if (h.getDetails() != null) { + p.addField("details", h.getDetails()); + } + if (h.getOperator() != null) { + p.addField("operator", h.getOperator()); + } + if (h.getRemark() != null) { + p.addField("remark", h.getRemark()); + } + return p; + } + + // ========== 查询 ========== + + @Override + public List selectByRecordId(Long alarmRecordId, Instant from, Instant to) { + // Known Pitfall 注入防护：alarmRecordId 是 Long，直接转 String 安全 + StringBuilder flux = buildBaseFlux(from, to); + flux.append(String.format( + " |> filter(fn: (r) => r._field == \"alarm_record_id\" and r._value == %d)\n", + alarmRecordId)); + flux.append(" |> sort(columns: [\"_time\"], desc: false)\n"); + return executeQuery(flux.toString()); + } + + @Override + public List selectByDeviceId(Long deviceId, Long tenantId, Instant from, Instant to) { + StringBuilder flux = buildBaseFlux(from, to); + flux.append(String.format( + " |> filter(fn: (r) => r.device_id == \"%d\" and r.tenant_id == \"%d\")\n", + deviceId, tenantId)); + flux.append(" |> sort(columns: [\"_time\"], desc: false)\n"); + return executeQuery(flux.toString()); + } + + @Override + public List queryLatestByDevice(Long deviceId, Long tenantId, int limit) { + StringBuilder flux = new StringBuilder(); + flux.append(String.format("from(bucket: \"%s\")\n", + FluxQuerySanitizer.escapeStringLiteral(properties.getBucket()))); + flux.append(" |> range(start: -365d)\n"); + flux.append(String.format( + " |> filter(fn: (r) => r._measurement == \"%s\")\n", + FluxQuerySanitizer.escapeStringLiteral(MEASUREMENT))); + flux.append(String.format( + " |> filter(fn: (r) => r.device_id == \"%d\" and r.tenant_id == \"%d\")\n", + deviceId, tenantId)); + flux.append(" |> sort(columns: [\"_time\"], desc: true)\n"); + flux.append(String.format(" |> limit(n: %d)\n", limit)); + return executeQuery(flux.toString()); + } + + @Override + public String getType() { + return "ctsdb"; + } + + // ========== 内部工具 ========== + + private StringBuilder buildBaseFlux(Instant from, Instant to) { + StringBuilder flux = new StringBuilder(); + flux.append(String.format("from(bucket: \"%s\")\n", + FluxQuerySanitizer.escapeStringLiteral(properties.getBucket()))); + if (from != null && to != null) { + flux.append(String.format(" |> range(start: %s, stop: %s)\n", + from.toString(), to.toString())); + } else { + flux.append(" |> range(start: -365d)\n"); + } + flux.append(String.format(" |> filter(fn: (r) => r._measurement == \"%s\")\n", + FluxQuerySanitizer.escapeStringLiteral(MEASUREMENT))); + return flux; + } + + private List executeQuery(String fluxQuery) { + try { + QueryApi queryApi = influxDBClient.getQueryApi(); + List tables = queryApi.query(fluxQuery); + List result = new ArrayList<>(); + for (FluxTable table : tables) { + for (FluxRecord record : table.getRecords()) { + result.add(recordToHistoryDO(record)); + } + } + return result; + } catch (Exception e) { + log.warn("[executeQuery][CTSDB 查询告警历史异常，返回空列表]: {}", e.getMessage()); + return Collections.emptyList(); + } + } + + private AlarmHistoryDO recordToHistoryDO(FluxRecord record) { + return AlarmHistoryDO.builder() + .ts(record.getTime()) + .deviceId(parseLong(record.getValueByKey("device_id"))) + .tenantId(parseLong(record.getValueByKey("tenant_id"))) + .alarmRecordId(parseLong(getFieldValue(record, "alarm_record_id"))) + .alarmConfigId(parseLong(getFieldValue(record, "alarm_config_id"))) + .severity(parseInt(getFieldValue(record, "severity"))) + .ackState(parseInt(getFieldValue(record, "ack_state"))) + .clearState(parseInt(getFieldValue(record, "clear_state"))) + .archived(parseLong(getFieldValue(record, "archived")) == 1L) + .eventType(strValue(getFieldValue(record, "event_type"))) + .triggerData(strValue(getFieldValue(record, "trigger_data"))) + .details(strValue(getFieldValue(record, "details"))) + .operator(strValue(getFieldValue(record, "operator"))) + .remark(strValue(getFieldValue(record, "remark"))) + .build(); + } + + private Object getFieldValue(FluxRecord record, String field) { + // FluxRecord 中 _field/_value 一行一个 field，tag 直接按 key 取 + String currentField = (String) record.getValueByKey("_field"); + if (field.equals(currentField)) { + return record.getValue(); + } + return record.getValueByKey(field); + } + + private static long nullSafe(Long v) { + return v != null ? v : 0L; + } + + private static long nullSafe(Integer v) { + return v != null ? v.longValue() : 0L; + } + + private static String strSafe(String v) { + return v != null ? v : ""; + } + + private static Long parseLong(Object v) { + if (v == null) return null; + if (v instanceof Number n) return n.longValue(); + try { return Long.parseLong(v.toString()); } catch (Exception e) { return null; } + } + + private static Integer parseInt(Object v) { + if (v == null) return null; + if (v instanceof Number n) return n.intValue(); + try { return Integer.parseInt(v.toString()); } catch (Exception e) { return null; } + } + + private static String strValue(Object v) { + return v != null ? v.toString() : null; + } + +} diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/tdengine/TdengineAlarmHistoryDaoImpl.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/tdengine/TdengineAlarmHistoryDaoImpl.java new file mode 100644 index 00000000..54344621 --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/tsdb/tdengine/TdengineAlarmHistoryDaoImpl.java @@ -0,0 +1,297 @@ +package com.viewsh.module.iot.dal.tsdb.tdengine; + +import com.viewsh.module.iot.dal.dataobject.alarm.AlarmHistoryDO; +import com.viewsh.module.iot.dal.tsdb.IotTsDbAlarmHistoryDao; +import jakarta.annotation.PostConstruct; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.RowMapper; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** + * TDengine 实现的告警历史时序 DAO + * + *

超级表（stable）结构： + * 

+ * CREATE STABLE alarm_history (
+ *     ts          TIMESTAMP,
+ *     alarm_record_id BIGINT,
+ *     alarm_config_id BIGINT,
+ *     severity    TINYINT,
+ *     ack_state   TINYINT,
+ *     clear_state TINYINT,
+ *     archived    TINYINT,
+ *     trigger_data NCHAR(2048),
+ *     details     NCHAR(2048),
+ *     operator    NCHAR(64),
+ *     remark      NCHAR(256),
+ *     event_type  NCHAR(32)
+ * ) TAGS (device_id BIGINT, tenant_id BIGINT);
+ *
+ *

+ * + *

Known Pitfall TDengine：按需动态创建子表 {@code alarm_history_{deviceId}}。

+ *

Known Pitfall F3：@PostConstruct 校验 retention（SHOW DATABASES），不 fail-fast。

+ *

Known Pitfall trigger_data：超过 2048 NCHAR 则截断（第一期 MVP，后续存 MinIO ref）。

+ * + * @author B13 + */ +@Slf4j +@RequiredArgsConstructor +public class TdengineAlarmHistoryDaoImpl implements IotTsDbAlarmHistoryDao { + + /** TDengine 子表名前缀 */ + private static final String SUB_TABLE_PREFIX = "alarm_history_"; + + /** 超级表名 */ + private static final String STABLE_NAME = "alarm_history"; + + /** trigger_data / details 的最大字符数（NCHAR 2048）*/ + private static final int MAX_NCHAR_LEN = 2048; + + /** operator 最大字符数 */ + private static final int MAX_OPERATOR_LEN = 64; + + /** remark 最大字符数 */ + private static final int MAX_REMARK_LEN = 256; + + /** event_type 最大字符数 */ + private static final int MAX_EVENT_TYPE_LEN = 32; + + /** + * TDengine 专用 JdbcTemplate（由 dynamic-datasource 路由到 tdengine 数据源） + * + *

注入时通过 {@code @TDengineDS} qualifier 选择 TDengine 连接池。 + * 若未配置 qualifier，可直接注入默认 JdbcTemplate 并在 datasource 路由层切换。

+ */ + private final JdbcTemplate jdbcTemplate; + + // ========== 评审 F3：启动时校验 retention ========== + + /** + * 校验 TDengine 的 KEEP 参数（retention）。 + *

不符合预期仅打 WARN，不 fail-fast。

+ */ + @PostConstruct + public void initSchema() { + // 1. 创建超级表（如果不存在） + try { + createStableIfAbsent(); + } catch (Exception e) { + log.warn("[initSchema][创建 TDengine alarm_history 超级表失败，请手动建表]: {}", e.getMessage()); + } + // 2. 校验 retention + verifyRetention(); + } + + private void createStableIfAbsent() { + String sql = "CREATE STABLE IF NOT EXISTS " + STABLE_NAME + " (" + + "ts TIMESTAMP, " + + "alarm_record_id BIGINT, " + + "alarm_config_id BIGINT, " + + "severity TINYINT, " + + "ack_state TINYINT, " + + "clear_state TINYINT, " + + "archived TINYINT, " + + "trigger_data NCHAR(2048), " + + "details NCHAR(2048), " + + "operator NCHAR(64), " + + "remark NCHAR(256), " + + "event_type NCHAR(32)" + + ") TAGS (device_id BIGINT, tenant_id BIGINT)"; + jdbcTemplate.execute(sql); + log.info("[createStableIfAbsent][TDengine alarm_history 超级表就绪]"); + } + + private void verifyRetention() { + try { + List rows = jdbcTemplate.query( + "SHOW DATABASES", + (rs, rowNum) -> rs.getString("name") + ":" + rs.getString("keep")); + // keep 一般格式为 "365,365,365"（对应三级存储天数），第一个值为热存 retention + for (String row : rows) { + if (row.startsWith("log:") || row.startsWith("information_schema:")) { + continue; // 跳过系统库 + } + String[] parts = row.split(":"); + if (parts.length == 2) { + String keepVal = parts[1].split(",")[0].trim(); + try { + int keepDays = Integer.parseInt(keepVal); + if (keepDays < 365) { + log.warn("[verifyRetention][TDengine 库 '{}' keep={}d 小于预期 365d，" + + "请检查 TDengine 数据库 KEEP 设置]", parts[0], keepDays); + } else { + log.info("[verifyRetention][TDengine 库 '{}' keep={}d 符合预期]", parts[0], keepDays); + } + } catch (NumberFormatException e) { + log.warn("[verifyRetention][无法解析 TDengine KEEP 值: {}]", keepVal); + } + } + } + } catch (Exception e) { + log.warn("[verifyRetention][校验 TDengine retention 失败，请检查连接配置]: {}", e.getMessage()); + } + } + + // ========== 写入 ========== + + @Override + public void insert(AlarmHistoryDO history) { + Objects.requireNonNull(history, "history 不能为空"); + ensureSubTable(history.getDeviceId(), history.getTenantId()); + String subTable = subTableName(history.getDeviceId()); + Instant ts = history.getTs() != null ? history.getTs() : Instant.now(); + + String sql = "INSERT INTO " + subTable + " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"; + jdbcTemplate.update(sql, + Timestamp.from(ts), + history.getAlarmRecordId(), + history.getAlarmConfigId(), + toTinyInt(history.getSeverity()), + toTinyInt(history.getAckState()), + toTinyInt(history.getClearState()), + history.getArchived() != null && history.getArchived() ? (byte) 1 : (byte) 0, + truncate(history.getTriggerData(), MAX_NCHAR_LEN), + truncate(history.getDetails(), MAX_NCHAR_LEN), + truncate(history.getOperator(), MAX_OPERATOR_LEN), + truncate(history.getRemark(), MAX_REMARK_LEN), + truncate(history.getEventType(), MAX_EVENT_TYPE_LEN) + ); + log.debug("[insert][TDengine 告警历史写入成功，alarmRecordId={}, eventType={}]", + history.getAlarmRecordId(), history.getEventType()); + } + + @Override + public void batchInsert(List list) { + if (list == null || list.isEmpty()) { + return; + } + // TDengine 批量：多行 INSERT INTO ... VALUES ...; ... 语法 + // 每个子表分组，减少子表创建开销 + for (AlarmHistoryDO history : list) { + insert(history); + } + } + + // ========== 查询 ========== + + @Override + public List selectByRecordId(Long alarmRecordId, Instant from, Instant to) { + try { + String sql = "SELECT * FROM " + STABLE_NAME + + " WHERE alarm_record_id = ? AND ts >= ? AND ts < ?" + + " AND tenant_id = alarm_history.tenant_id" + // 多租户隔离 + " ORDER BY ts ASC"; + // 简化：直接用时间范围过滤，并按 alarm_record_id 过滤 + String querySql = "SELECT * FROM " + STABLE_NAME + + " WHERE alarm_record_id = ?" + + (from != null ? " AND ts >= " + from.toEpochMilli() : "") + + (to != null ? " AND ts < " + to.toEpochMilli() : "") + + " ORDER BY ts ASC"; + return jdbcTemplate.query(querySql, new AlarmHistoryRowMapper(), alarmRecordId); + } catch (Exception e) { + log.warn("[selectByRecordId][TDengine 查询失败，返回空列表]: {}", e.getMessage()); + return Collections.emptyList(); + } + } + + @Override + public List selectByDeviceId(Long deviceId, Long tenantId, Instant from, Instant to) { + try { + String subTable = subTableName(deviceId); + StringBuilder sql = new StringBuilder("SELECT * FROM ").append(subTable).append(" WHERE 1=1"); + if (from != null) { + sql.append(" AND ts >= ").append(from.toEpochMilli()); + } + if (to != null) { + sql.append(" AND ts < ").append(to.toEpochMilli()); + } + sql.append(" ORDER BY ts ASC"); + return jdbcTemplate.query(sql.toString(), new AlarmHistoryRowMapper()); + } catch (Exception e) { + log.warn("[selectByDeviceId][TDengine 查询失败，返回空列表，deviceId={}]: {}", deviceId, e.getMessage()); + return Collections.emptyList(); + } + } + + @Override + public List queryLatestByDevice(Long deviceId, Long tenantId, int limit) { + try { + String subTable = subTableName(deviceId); + String sql = "SELECT * FROM " + subTable + " ORDER BY ts DESC LIMIT " + limit; + return jdbcTemplate.query(sql, new AlarmHistoryRowMapper()); + } catch (Exception e) { + log.warn("[queryLatestByDevice][TDengine 查询失败，返回空列表，deviceId={}]: {}", deviceId, e.getMessage()); + return Collections.emptyList(); + } + } + + @Override + public String getType() { + return "tdengine"; + } + + // ========== 内部工具 ========== + + /** 按需创建子表（第一次插入该 device_id 时执行，IF NOT EXISTS 保证幂等） */ + private void ensureSubTable(Long deviceId, Long tenantId) { + String subTable = subTableName(deviceId); + String sql = "CREATE TABLE IF NOT EXISTS " + subTable + + " USING " + STABLE_NAME + + " TAGS (" + nullSafe(deviceId) + ", " + nullSafe(tenantId) + ")"; + jdbcTemplate.execute(sql); + } + + private static String subTableName(Long deviceId) { + return SUB_TABLE_PREFIX + nullSafe(deviceId); + } + + private static long nullSafe(Long v) { + return v != null ? v : 0L; + } + + private static byte toTinyInt(Integer v) { + return v != null ? v.byteValue() : 0; + } + + private static String truncate(String v, int maxLen) { + if (v == null) return null; + return v.length() > maxLen ? v.substring(0, maxLen) : v; + } + + // ========== RowMapper ========== + + private static class AlarmHistoryRowMapper implements RowMapper { + @Override + public AlarmHistoryDO mapRow(ResultSet rs, int rowNum) throws SQLException { + Timestamp ts = rs.getTimestamp("ts"); + return AlarmHistoryDO.builder() + .ts(ts != null ? ts.toInstant() : null) + .alarmRecordId(rs.getObject("alarm_record_id", Long.class)) + .alarmConfigId(rs.getObject("alarm_config_id", Long.class)) + .deviceId(rs.getObject("device_id", Long.class)) + .tenantId(rs.getObject("tenant_id", Long.class)) + .severity(rs.getObject("severity") != null ? rs.getInt("severity") : null) + .ackState(rs.getObject("ack_state") != null ? rs.getInt("ack_state") : null) + .clearState(rs.getObject("clear_state") != null ? rs.getInt("clear_state") : null) + .archived(rs.getObject("archived") != null && rs.getByte("archived") == 1) + .triggerData(rs.getString("trigger_data")) + .details(rs.getString("details")) + .operator(rs.getString("operator")) + .remark(rs.getString("remark")) + .eventType(rs.getString("event_type")) + .build(); + } + } + +} diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/framework/tsdb/config/TsDbAutoConfiguration.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/framework/tsdb/config/TsDbAutoConfiguration.java index cc0f831e..1ef52a89 100644 --- a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/framework/tsdb/config/TsDbAutoConfiguration.java +++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/framework/tsdb/config/TsDbAutoConfiguration.java @@ -2,6 +2,9 @@ package com.viewsh.module.iot.framework.tsdb.config; import com.viewsh.module.iot.dal.tdengine.IotDeviceMessageMapper; import com.viewsh.module.iot.dal.tdengine.IotDevicePropertyMapper; +import com.viewsh.module.iot.dal.tsdb.IotTsDbAlarmHistoryDao; +import com.viewsh.module.iot.dal.tsdb.ctsdb.CtsdbAlarmHistoryDaoImpl; +import com.viewsh.module.iot.dal.tsdb.tdengine.TdengineAlarmHistoryDaoImpl; import com.viewsh.module.iot.framework.tsdb.IotTsDbDeviceMessageDao; import com.viewsh.module.iot.framework.tsdb.IotTsDbDevicePropertyDao; import com.viewsh.module.iot.framework.tsdb.tdengine.TDengineDeviceMessageDaoImpl; @@ -13,6 +16,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; /** * 时序数据库自动装配 @@ -53,6 +57,19 @@ public class TsDbAutoConfiguration { return new TDengineDevicePropertyDaoImpl(mapper); } + /** + * [B13] TDengine 告警历史 DAO + * + *

注意：jdbcTemplate 需通过动态数据源路由到 TDengine 数据源。 + * 在 dynamic-datasource 环境中，可通过 {@code @DS("tdengine")} 在 Service 方法上切换， + * 或注入专用的 TDengine JdbcTemplate Bean。

+ */ + @Bean + @ConditionalOnProperty(name = "viewsh.iot.tsdb.type", havingValue = "tdengine", matchIfMissing = true) + public IotTsDbAlarmHistoryDao tdengineAlarmHistoryDao(JdbcTemplate jdbcTemplate) { + return new TdengineAlarmHistoryDaoImpl(jdbcTemplate); + } + // ========== CTSDB (InfluxDB) 实现 ========== // 使用内部类 + @ConditionalOnClass 保护，避免 classpath 无 InfluxDB 依赖时 NoClassDefFoundError @@ -79,6 +96,16 @@ public class TsDbAutoConfiguration { influxDBClient, properties, ctsdbWriteApi); } + /** + * [B13] CTSDB 告警历史 DAO + */ + @Bean + public IotTsDbAlarmHistoryDao ctsdbAlarmHistoryDao( + com.influxdb.client.InfluxDBClient influxDBClient, + com.viewsh.module.iot.framework.tsdb.ctsdb.CtsdbProperties properties) { + return new CtsdbAlarmHistoryDaoImpl(influxDBClient, properties); + } + } } diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/alarm/IotAlarmHistoryService.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/alarm/IotAlarmHistoryService.java new file mode 100644 index 00000000..22ab59f1 --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/alarm/IotAlarmHistoryService.java @@ -0,0 +1,148 @@ +package com.viewsh.module.iot.service.alarm; + +import com.viewsh.module.iot.dal.dataobject.alarm.AlarmHistoryDO; +import com.viewsh.module.iot.dal.tsdb.IotTsDbAlarmHistoryDao; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.time.Instant; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** + * IoT 告警历史 Service + * + *

协调对时序数据库（CTSDB 或 TDengine）的告警历史读写，屏蔽底层实现切换。

+ * + *

Known Pitfall F1：告警历史是审计数据，insert 同步写入（不走 PersistenceBuffer 异步缓冲）， + * 避免进程崩溃丢失审计记录。批量操作（如批量归档）可以走调用方异步。

+ * + *

Known Pitfall F1 降级：TSDB 写失败不影响主流程——记录日志，不重新抛出。 + * 查询失败返回空列表 + log WARN。

+ * + *

通过 {@code viewsh.iot.tsdb.type} 选主实现：

+ *
    + *
  • {@code ctsdb} → {@link com.viewsh.module.iot.dal.tsdb.ctsdb.CtsdbAlarmHistoryDaoImpl}
    • + *
  • {@code tdengine}→ {@link com.viewsh.module.iot.dal.tsdb.tdengine.TdengineAlarmHistoryDaoImpl}
    • + *
+ * + * @author B13 + */ +@Slf4j +@Service +public class IotAlarmHistoryService { + + private final IotTsDbAlarmHistoryDao dao; + + /** + * 通过 Spring 注入已激活的 {@link IotTsDbAlarmHistoryDao} 实现（由 @ConditionalOnProperty 决定）。 + * + *

若未来需要多实现并存，可改为 {@code List} 注入并用 type 路由。

+ */ + public IotAlarmHistoryService(IotTsDbAlarmHistoryDao dao) { + this.dao = dao; + } + + /** + * 记录一条告警历史（同步写入 TSDB） + * + *

TSDB 写失败不抛出异常，仅记录 ERROR 日志，保证主流程不受影响。

+ * + * @param history 告警历史 DO + */ + public void record(AlarmHistoryDO history) { + Objects.requireNonNull(history, "history 不能为空"); + if (history.getTs() == null) { + history.setTs(Instant.now()); + } + try { + dao.insert(history); + } catch (Exception e) { + // Known Pitfall F1：TSDB 写失败不影响主流程 + log.error("[record][告警历史写入 TSDB 失败，alarmRecordId={}, eventType={}]: {}", + history.getAlarmRecordId(), history.getEventType(), e.getMessage(), e); + } + } + + /** + * 批量记录告警历史 + * + * @param list 告警历史列表 + */ + public void batchRecord(List list) { + if (list == null || list.isEmpty()) { + return; + } + try { + dao.batchInsert(list); + } catch (Exception e) { + log.error("[batchRecord][批量告警历史写入 TSDB 失败，count={}]: {}", list.size(), e.getMessage(), e); + } + } + + /** + * 按告警记录 ID 查询历史（告警详情页） + * + *

TSDB 不可用时返回空列表 + log WARN。

+ * + * @param alarmRecordId 告警记录 ID + * @param from 开始时间（null 表示不限） + * @param to 结束时间（null 表示不限） + * @return 历史列表，按时间 ASC；TSDB 异常时返回空列表 + */ + public List queryByAlarmRecord(Long alarmRecordId, Instant from, Instant to) { + try { + return dao.selectByRecordId(alarmRecordId, from, to); + } catch (Exception e) { + log.warn("[queryByAlarmRecord][查询告警历史失败，alarmRecordId={}，返回空列表]: {}", + alarmRecordId, e.getMessage()); + return Collections.emptyList(); + } + } + + /** + * 按设备查询历史（告警趋势） + * + * @param deviceId 设备 ID + * @param tenantId 租户 ID + * @param from 开始时间 + * @param to 结束时间 + * @return 历史列表，按时间 ASC；TSDB 异常时返回空列表 + */ + public List queryByDevice(Long deviceId, Long tenantId, Instant from, Instant to) { + try { + return dao.selectByDeviceId(deviceId, tenantId, from, to); + } catch (Exception e) { + log.warn("[queryByDevice][查询告警历史失败，deviceId={}，返回空列表]: {}", deviceId, e.getMessage()); + return Collections.emptyList(); + } + } + + /** + * 按设备查询最近 N 条（快速趋势视图） + * + * @param deviceId 设备 ID + * @param tenantId 租户 ID + * @param limit 条数上限 + * @return 历史列表；TSDB 异常时返回空列表 + */ + public List queryLatestByDevice(Long deviceId, Long tenantId, int limit) { + try { + return dao.queryLatestByDevice(deviceId, tenantId, limit); + } catch (Exception e) { + log.warn("[queryLatestByDevice][查询告警历史失败，deviceId={}，返回空列表]: {}", deviceId, e.getMessage()); + return Collections.emptyList(); + } + } + + /** + * 返回当前激活的 TSDB 实现类型（供监控/健康检查用） + * + * @return "ctsdb" 或 "tdengine" + */ + public String getTsdbType() { + return dao.getType(); + } + +} diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/test/java/com/viewsh/module/iot/dal/tsdb/AlarmHistoryDaoTest.java b/viewsh-module-iot/viewsh-module-iot-server/src/test/java/com/viewsh/module/iot/dal/tsdb/AlarmHistoryDaoTest.java new file mode 100644 index 00000000..cb3f5e47 --- /dev/null +++ b/viewsh-module-iot/viewsh-module-iot-server/src/test/java/com/viewsh/module/iot/dal/tsdb/AlarmHistoryDaoTest.java @@ -0,0 +1,203 @@ +package com.viewsh.module.iot.dal.tsdb; + +import com.influxdb.client.InfluxDBClient; +import com.influxdb.client.QueryApi; +import com.influxdb.client.WriteApiBlocking; +import com.influxdb.client.domain.WritePrecision; +import com.influxdb.client.write.Point; +import com.viewsh.module.iot.dal.dataobject.alarm.AlarmHistoryDO; +import com.viewsh.module.iot.dal.tsdb.ctsdb.CtsdbAlarmHistoryDaoImpl; +import com.viewsh.module.iot.dal.tsdb.tdengine.TdengineAlarmHistoryDaoImpl; +import com.viewsh.module.iot.framework.tsdb.ctsdb.CtsdbProperties; +import com.viewsh.module.iot.service.alarm.IotAlarmHistoryService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.time.Instant; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.contains; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.*; + +/** + * B13 AlarmHistory 时序表 DAO 单元测试（5 个用例） + * + *

全部使用 Mockito，不依赖真实 CTSDB/TDengine，适合 CI 快速验证。

+ * + * @author B13 + */ +@ExtendWith(MockitoExtension.class) +class AlarmHistoryDaoTest { + + // ========== CTSDB mock ========== + @Mock + private InfluxDBClient influxDBClient; + + @Mock + private WriteApiBlocking writeApiBlocking; + + @Mock + private QueryApi queryApi; + + // ========== TDengine mock ========== + @Mock + private JdbcTemplate jdbcTemplate; + + private CtsdbProperties ctsdbProperties; + + private CtsdbAlarmHistoryDaoImpl ctsdbDao; + private TdengineAlarmHistoryDaoImpl tdengineDao; + + @BeforeEach + void setUp() { + ctsdbProperties = new CtsdbProperties(); + ctsdbProperties.setBucket("aiot_platform"); + ctsdbProperties.setOrg("aiot"); + ctsdbProperties.setUrl("http://localhost:8086"); + + // 使用构造函数直接创建（不走 Spring，跳过 @PostConstruct 的 @Bean 注册） + ctsdbDao = new CtsdbAlarmHistoryDaoImpl(influxDBClient, ctsdbProperties) { + @Override + public void verifyRetention() { + // 测试中跳过 retention 校验（避免真实 InfluxDB 调用） + } + }; + + tdengineDao = new TdengineAlarmHistoryDaoImpl(jdbcTemplate) { + @Override + public void initSchema() { + // 测试中跳过建表和 retention 校验 + } + }; + } + + // ==================== 用例 1：ctsdb_insert_success ==================== + + @Test + @DisplayName("用例 1：CTSDB 模式写入成功 — Point 写入 bucket") + void ctsdb_insert_success() { + // Arrange + when(influxDBClient.getWriteApiBlocking()).thenReturn(writeApiBlocking); + AlarmHistoryDO history = buildHistory("trigger"); + + // Act + ctsdbDao.insert(history); + + // Assert：writePoint 被调用 + verify(writeApiBlocking, times(1)) + .writePoint(eq("aiot_platform"), eq("aiot"), any(Point.class)); + } + + // ==================== 用例 2：tdengine_insert_success ==================== + + @Test + @DisplayName("用例 2：TDengine 模式写入成功 — 子表 alarm_history_{did} 写入记录") + void tdengine_insert_success() { + // Arrange — 模拟 CREATE TABLE IF NOT EXISTS 和 INSERT 成功 + doNothing().when(jdbcTemplate).execute(anyString()); + when(jdbcTemplate.update(anyString(), any(Object[].class))).thenReturn(1); + + AlarmHistoryDO history = buildHistory("trigger"); + + // Act + tdengineDao.insert(history); + + // Assert：子表创建 SQL 包含正确前缀 + ArgumentCaptor sqlCaptor = ArgumentCaptor.forClass(String.class); + verify(jdbcTemplate, atLeastOnce()).execute(sqlCaptor.capture()); + boolean hasSubTableCreate = sqlCaptor.getAllValues().stream() + .anyMatch(sql -> sql.contains("alarm_history_100")); + assertThat(hasSubTableCreate).isTrue(); + + // Assert：INSERT 被调用 + verify(jdbcTemplate, times(1)).update(contains("alarm_history_100"), any(Object[].class)); + } + + // ==================== 用例 3：service_insert_sync ==================== + + @Test + @DisplayName("用例 3：AlarmHistoryService.record 调用后 DAO.insert 被同步调用") + void service_insert_sync() { + // Arrange：使用 mock DAO + IotTsDbAlarmHistoryDao mockDao = mock(IotTsDbAlarmHistoryDao.class); + IotAlarmHistoryService service = new IotAlarmHistoryService(mockDao); + + AlarmHistoryDO history = buildHistory("ack"); + + // Act + service.record(history); + + // Assert：DAO.insert 被同步调用一次（F1：审计数据同步写入） + verify(mockDao, times(1)).insert(history); + } + + // ==================== 用例 4：service_tsdb_failure_nothrow ==================== + + @Test + @DisplayName("用例 4：TSDB 写失败，service 不抛异常（主流程不受影响）") + void service_tsdb_failure_nothrow() { + // Arrange：DAO.insert 抛出运行时异常 + IotTsDbAlarmHistoryDao failingDao = mock(IotTsDbAlarmHistoryDao.class); + doThrow(new RuntimeException("TDengine 连接超时")) + .when(failingDao).insert(any(AlarmHistoryDO.class)); + IotAlarmHistoryService service = new IotAlarmHistoryService(failingDao); + + AlarmHistoryDO history = buildHistory("clear"); + + // Act + Assert：不应抛出任何异常 + assertThatCode(() -> service.record(history)) + .doesNotThrowAnyException(); + } + + // ==================== 用例 5：query_returns_empty_on_error ==================== + + @Test + @DisplayName("用例 5：TSDB 查询异常，返回空列表") + void query_returns_empty_on_error() { + // Arrange：DAO.selectByRecordId 抛出异常 + IotTsDbAlarmHistoryDao failingDao = mock(IotTsDbAlarmHistoryDao.class); + when(failingDao.selectByRecordId(anyLong(), any(), any())) + .thenThrow(new RuntimeException("CTSDB 查询超时")); + IotAlarmHistoryService service = new IotAlarmHistoryService(failingDao); + + // Act + List result = service.queryByAlarmRecord(1L, null, null); + + // Assert + assertThat(result).isNotNull().isEmpty(); + } + + // ==================== 辅助方法 ==================== + + private static AlarmHistoryDO buildHistory(String eventType) { + return AlarmHistoryDO.builder() + .ts(Instant.now()) + .alarmRecordId(1L) + .alarmConfigId(10L) + .deviceId(100L) + .tenantId(1L) + .severity(3) + .ackState(0) + .clearState(0) + .archived(false) + .eventType(eventType) + .triggerData("{\"value\":42}") + .details("{\"msg\":\"over threshold\"}") + .operator("admin") + .remark("test remark") + .build(); + } + +}