feat(aigc): 面五 game-cloud 生成业务埋点 GenMetrics(观测阶段四·flag 默认关)
新增 GenMetrics 组件(@ConditionalOnProperty huijing.aigc.metrics.enabled,默认关=零装配零副作用),
把 MVP 要盯的生成指标以 micrometer 暴露给已活的 /actuator/prometheus:
- gen_task_total{status} Counter:终态计成功率(succeeded/failed/timed_out,超时据真实 failureReason 还原)
- gen_queue_depth Gauge:@Scheduled 定时刷 countGlobalInflight→内存 AtomicInteger,scrape 零 DB(勿全表重查)
- gen_gate_fail_total{gate} Counter:九门逐门失败,源 trace.sevenGateVerdict.guards(复用 ReadinessScorer.guardPass:H_progress 嵌套取 .pass)
- gen_duration_seconds Timer(直方图):源 trace.wallS
- llm_cost DistributionSummary:源 trace.cost.totalRmb(worker token×倍率自估值、非 new-api 权威计费)
埋点挂 DifyCallbackServiceImpl.postCommitBestEffort(post-commit best-effort、realSuccess 判定前、
覆盖成功/失败/超时三类终态),null 判空 + 内部吞异常双保险,绝不进 DifyCallbackTxService 事务方法、
不咬回调主链(§8 旁路铁律)。micrometer-core 经 spring-ai-commons 传递,无需改 pom。
GenMetricsTest 7 例(终态三标签/耗时成本/逐门guards嵌套裸/队列gauge/脏trace吞/开关装配三态/抛错不咬主链)
+ 主链回归 DifyCallbackServiceImplTest 20 + DifyCallbackTxTraceTest 6,主控亲跑 BUILD SUCCESS 全绿。
面一 actuator 已活;dev 重部署后 profile 开 flag 即出指标(下一窗口深冒烟真验)。
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lili
parent
dee9158504
commit
7e21b331a5
@ -0,0 +1,257 @@
|
|||||||
|
package com.wanxiang.huijing.game.module.aigc.framework.metrics;
|
||||||
|
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.controller.admin.task.vo.DifyCallbackReqVO;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.dal.dataobject.task.AigcTaskDO;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.dal.mysql.task.AigcTaskMapper;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.enums.AigcTaskStatusEnum;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.enums.FailureReasonEnum;
|
||||||
|
import com.wanxiang.huijing.framework.tenant.core.util.TenantUtils;
|
||||||
|
import io.micrometer.core.instrument.DistributionSummary;
|
||||||
|
import io.micrometer.core.instrument.Gauge;
|
||||||
|
import io.micrometer.core.instrument.MeterRegistry;
|
||||||
|
import io.micrometer.core.instrument.Timer;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||||
|
import org.springframework.scheduling.annotation.Scheduled;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 生成业务语义埋点（观测体系阶段四 · 面五）
|
||||||
|
*
|
||||||
|
* <p>把 MVP 真正要盯的几个生成指标以 micrometer 形式暴露给 {@code /actuator/prometheus}：生成成功率、单次耗时、
|
||||||
|
* 单次成本、九门逐门失败、在飞队列深度。这些值此前只落在产物目录的 json 或回调的 trace 字段里，没有一处能实时看、
|
||||||
|
* 能告警——本组件是把它们变成可采集指标的收口。
|
||||||
|
*
|
||||||
|
* <p><b>开关与装配（旁路铁律）</b>：整组埋点由 {@code huijing.aigc.metrics.enabled} 控制，<b>默认关</b>——
|
||||||
|
* {@link ConditionalOnProperty} 保证关时本组件根本不装配（无 Bean、无 @Scheduled、无 meter 注册），
|
||||||
|
* 对生成主链零字节影响。生产 profile 显式置 true 才开。开启的前提是「面一」已给单体引入 actuator + micrometer、
|
||||||
|
* 提供 {@link MeterRegistry} bean；本组件只消费该 bean、不负责装 actuator。故未做面一的环境保持开关关即可。
|
||||||
|
*
|
||||||
|
* <p><b>埋点绝不咬主链（硬红线）</b>：{@link #recordTerminal} 全程吞异常，调用方（{@code DifyCallbackServiceImpl}
|
||||||
|
* 的提交后 best-effort 段）再兜一层，双保险确保任何埋点故障都只留痕、绝不外抛去影响回调的成败判定或后续通知/回填。
|
||||||
|
*
|
||||||
|
* <p><b>Micrometer 点名 → Prometheus 名</b>（PrometheusMeterRegistry 自动转 snake_case 并补后缀）：
|
||||||
|
* <ul>
|
||||||
|
* <li>{@code gen.task}{status} → {@code gen_task_total{status}}（Counter，成功率 = succeeded / 全部）</li>
|
||||||
|
* <li>{@code gen.gate.fail}{gate} → {@code gen_gate_fail_total{gate}}（Counter，九门逐门失败）</li>
|
||||||
|
* <li>{@code gen.queue.depth} → {@code gen_queue_depth}（Gauge，在飞任务数）</li>
|
||||||
|
* <li>{@code gen.duration} → {@code gen_duration_seconds}（Timer，基准单位秒、直方图供 P50/P95）</li>
|
||||||
|
* <li>{@code llm.cost} → {@code llm_cost}（DistributionSummary，单次成本自估值）</li>
|
||||||
|
* </ul>
|
||||||
|
*
|
||||||
|
* @author 绘境AI（观测体系阶段四 · 面五 game-cloud 业务埋点）
|
||||||
|
*/
|
||||||
|
@Slf4j
|
||||||
|
@Component
|
||||||
|
@ConditionalOnProperty(name = "huijing.aigc.metrics.enabled", havingValue = "true")
|
||||||
|
public class GenMetrics {
|
||||||
|
|
||||||
|
/** gen_task_total{status}：任务终态计数（成功率分母 = 各 status 之和）。 */
|
||||||
|
static final String M_TASK = "gen.task";
|
||||||
|
/** gen_gate_fail_total{gate}：九门逐门失败计数。 */
|
||||||
|
static final String M_GATE_FAIL = "gen.gate.fail";
|
||||||
|
/** gen_queue_depth：当前在飞（queued+running）任务数。 */
|
||||||
|
static final String M_QUEUE_DEPTH = "gen.queue.depth";
|
||||||
|
/** gen_duration_seconds：单次生成端到端耗时。 */
|
||||||
|
static final String M_DURATION = "gen.duration";
|
||||||
|
/** llm_cost：单次生成成本（自估值）。 */
|
||||||
|
static final String M_LLM_COST = "llm.cost";
|
||||||
|
|
||||||
|
private final MeterRegistry registry;
|
||||||
|
private final AigcTaskMapper aigcTaskMapper;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* gen_queue_depth 的承载值：{@link #refreshQueueDepth} 定时刷、Gauge 只读它。
|
||||||
|
* 之所以不把 Gauge 直接绑到 {@code countGlobalInflight}，见 {@link #refreshQueueDepth} 注释。
|
||||||
|
*/
|
||||||
|
private final AtomicInteger queueDepth = new AtomicInteger(0);
|
||||||
|
|
||||||
|
/** gen_duration_seconds：构造期注册一次、复用（源=worker 侧算好的 trace.wallS）。 */
|
||||||
|
private final Timer genDuration;
|
||||||
|
|
||||||
|
/** llm_cost：构造期注册一次、复用（源=worker 侧 trace.cost.totalRmb 自估值）。 */
|
||||||
|
private final DistributionSummary llmCost;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param registry micrometer 指标注册表（面一装 actuator+micrometer 后在席；开关开时必需）
|
||||||
|
* @param aigcTaskMapper 任务 Mapper（复用其既有全局在飞计数 {@code countGlobalInflight}，不新增查询）
|
||||||
|
*/
|
||||||
|
public GenMetrics(MeterRegistry registry, AigcTaskMapper aigcTaskMapper) {
|
||||||
|
this.registry = registry;
|
||||||
|
this.aigcTaskMapper = aigcTaskMapper;
|
||||||
|
// Gauge 绑到内存 AtomicInteger：scrape 时零 DB，真实计数由 @Scheduled 定时刷入（见 refreshQueueDepth）。
|
||||||
|
Gauge.builder(M_QUEUE_DEPTH, queueDepth, AtomicInteger::doubleValue)
|
||||||
|
.description("当前在飞（queued+running）生成任务数")
|
||||||
|
.register(registry);
|
||||||
|
// Timer 基准单位为秒：meter 名故意不带 seconds，由 Prometheus 渲染补 _seconds 后缀（否则会双后缀）。
|
||||||
|
this.genDuration = Timer.builder(M_DURATION)
|
||||||
|
.description("单次生成端到端耗时（源=worker 侧 trace.wallS）")
|
||||||
|
.publishPercentileHistogram() // 产 _bucket，供 Prometheus 跨实例聚合 P50/P95
|
||||||
|
.register(registry);
|
||||||
|
this.llmCost = DistributionSummary.builder(M_LLM_COST)
|
||||||
|
.description("单次生成成本¥——worker 按 token×倍率的自估值，非 new-api 权威计费流水")
|
||||||
|
.register(registry);
|
||||||
|
log.info("[GenMetrics] 生成业务埋点已装配（huijing.aigc.metrics.enabled=true）");
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 终态收口埋点：回调事务提交后由 best-effort 段调用，一次埋全 gen_task_total / gen_duration / llm_cost / gen_gate_fail。
|
||||||
|
*
|
||||||
|
* <p>覆盖全部终态——成功、业务失败、以及超时收尸（stale RUNNING 收尸为 failed+failureReason=timeout）均经
|
||||||
|
* 回调唯一写入路径到达提交后段，故此处一个挂点即全覆盖，无需另挂 watchdog。
|
||||||
|
*
|
||||||
|
* <p><b>全程吞异常</b>：埋点是纯旁路，任何解析/注册异常只留痕、绝不外抛（旁路铁律）。
|
||||||
|
*
|
||||||
|
* @param reqVO 回调入参（取 trace 载荷：wallS / cost.totalRmb / sevenGateVerdict.guards）
|
||||||
|
* @param task 提交后按 traceId 回查到的任务（取终态 status + failureReason；可能为 null=未定位到）
|
||||||
|
*/
|
||||||
|
public void recordTerminal(DifyCallbackReqVO reqVO, AigcTaskDO task) {
|
||||||
|
try {
|
||||||
|
recordTaskTerminal(task);
|
||||||
|
Map<String, Object> trace = reqVO == null ? null : reqVO.getTrace();
|
||||||
|
if (trace != null && !trace.isEmpty()) {
|
||||||
|
recordDuration(trace);
|
||||||
|
recordCost(trace);
|
||||||
|
recordGateFails(trace);
|
||||||
|
}
|
||||||
|
} catch (Exception e) {
|
||||||
|
// 埋点整体兜底：任何异常只留痕，绝不外抛（回调主链不受影响）
|
||||||
|
log.error("[GenMetrics] 终态埋点异常（best-effort，不影响回调主链）traceId={}",
|
||||||
|
reqVO == null ? null : reqVO.getTraceId(), e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** gen_task_total{status}：按任务终态计数。 */
|
||||||
|
private void recordTaskTerminal(AigcTaskDO task) {
|
||||||
|
if (task == null || task.getStatus() == null) {
|
||||||
|
return; // 回调未定位到任务 → 无终态可计（属异常情形，另有 warn 留痕在回调侧）
|
||||||
|
}
|
||||||
|
String statusLabel = resolveStatusLabel(task);
|
||||||
|
if (statusLabel != null) {
|
||||||
|
registry.counter(M_TASK, "status", statusLabel).increment();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 终态标签口径（对齐设计 §6 succeeded / failed / timed_out 三分）：
|
||||||
|
* <ul>
|
||||||
|
* <li>SUCCEEDED(2) → {@code succeeded}</li>
|
||||||
|
* <li>FAILED(3) 且 failureReason=timeout → {@code timed_out}：系统状态机不写 status=4（§1.3-7 回调门禁只 succeeded/failed
|
||||||
|
* 两值，超时收尸落 failed+timeout），此处按<b>真实 failureReason</b> 把超时从 failed 里还原成 timed_out——据实映射、非编造</li>
|
||||||
|
* <li>FAILED(3) 其余 → {@code failed}</li>
|
||||||
|
* <li>其余终态（canceled 等）→ 枚举名小写（防御，正常回调路不产 canceled）</li>
|
||||||
|
* </ul>
|
||||||
|
*/
|
||||||
|
private String resolveStatusLabel(AigcTaskDO task) {
|
||||||
|
AigcTaskStatusEnum st = AigcTaskStatusEnum.of(task.getStatus());
|
||||||
|
if (st == null) {
|
||||||
|
return null; // 非法状态值：不计（避免脏标签污染指标基数）
|
||||||
|
}
|
||||||
|
if (st == AigcTaskStatusEnum.FAILED
|
||||||
|
&& FailureReasonEnum.TIMEOUT.getReason().equals(task.getFailureReason())) {
|
||||||
|
return "timed_out";
|
||||||
|
}
|
||||||
|
return st.name().toLowerCase();
|
||||||
|
}
|
||||||
|
|
||||||
|
/** gen_duration_seconds：trace.wallS（秒，worker 已算）→ Timer.record。缺失/非数/负值跳过。 */
|
||||||
|
private void recordDuration(Map<String, Object> trace) {
|
||||||
|
Object wallS = trace.get("wallS");
|
||||||
|
if (wallS instanceof Number) {
|
||||||
|
double sec = ((Number) wallS).doubleValue();
|
||||||
|
if (sec >= 0) {
|
||||||
|
// wallS 是秒级浮点（生成通常上百秒），按毫秒记入 Timer，精度足够。
|
||||||
|
genDuration.record((long) (sec * 1000), TimeUnit.MILLISECONDS);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* llm_cost：trace.cost.totalRmb → DistributionSummary.record（sum=累计花费、count=生成笔数）。
|
||||||
|
* 值是 worker 按 token×倍率的自估值，非 new-api 权威计费流水（取价失败降级时无 totalRmb，此处自然跳过）。
|
||||||
|
*/
|
||||||
|
@SuppressWarnings("unchecked")
|
||||||
|
private void recordCost(Map<String, Object> trace) {
|
||||||
|
Object cost = trace.get("cost");
|
||||||
|
if (cost instanceof Map) {
|
||||||
|
Object totalRmb = ((Map<String, Object>) cost).get("totalRmb");
|
||||||
|
if (totalRmb instanceof Number) {
|
||||||
|
double rmb = ((Number) totalRmb).doubleValue();
|
||||||
|
if (rmb >= 0) {
|
||||||
|
llmCost.record(rmb);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* gen_gate_fail_total{gate}：九门逐门失败计数，源 = trace.sevenGateVerdict.guards（{门名: pass}）。
|
||||||
|
*
|
||||||
|
* <p>guards 在便宜档主路与 tier2 两条生成线都逐门产出（result_out / service._extract_trace 同源），故本项可做到真逐门，
|
||||||
|
* 不必降级为合成分。门判真值解析复用 {@code ReadinessScorer.guardPass} 同款口径：H_progress 是嵌套对象
|
||||||
|
* {@code {pass,checks,latch,...}} 取其 .pass，其余门为裸 bool。只有明确「未过」的门才 +1。
|
||||||
|
*/
|
||||||
|
@SuppressWarnings("unchecked")
|
||||||
|
private void recordGateFails(Map<String, Object> trace) {
|
||||||
|
Object verdict = trace.get("sevenGateVerdict");
|
||||||
|
if (!(verdict instanceof Map)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
Object guards = ((Map<String, Object>) verdict).get("guards");
|
||||||
|
if (!(guards instanceof Map)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
for (Map.Entry<String, Object> gate : ((Map<String, Object>) guards).entrySet()) {
|
||||||
|
if (!guardPass(gate.getValue())) {
|
||||||
|
registry.counter(M_GATE_FAIL, "gate", gate.getKey()).increment();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 门判真值（对齐 ReadinessScorer.guardPass）：嵌套对象取 .pass；裸 bool/数字/字符串宽松判真。 */
|
||||||
|
@SuppressWarnings("unchecked")
|
||||||
|
private static boolean guardPass(Object gate) {
|
||||||
|
if (gate instanceof Map) {
|
||||||
|
return asBool(((Map<String, Object>) gate).get("pass"));
|
||||||
|
}
|
||||||
|
return asBool(gate);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 宽松布尔解析（trace 来自 JSON 反序列化，类型可能为 Boolean/String/Number）。 */
|
||||||
|
private static boolean asBool(Object v) {
|
||||||
|
if (v instanceof Boolean) {
|
||||||
|
return (Boolean) v;
|
||||||
|
}
|
||||||
|
if (v instanceof Number) {
|
||||||
|
return ((Number) v).doubleValue() != 0;
|
||||||
|
}
|
||||||
|
return v != null && "true".equalsIgnoreCase(String.valueOf(v));
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* gen_queue_depth 定时刷新（默认 15s，可配 {@code huijing.aigc.metrics.queue-depth-refresh-ms}）。
|
||||||
|
*
|
||||||
|
* <p><b>为何不把 Gauge 直接绑 {@code countGlobalInflight}</b>：直接绑会让每次 Prometheus scrape（以及 micrometer
|
||||||
|
* 内部 gauge 轮询）都打一次全表 COUNT，DB 压力与 scrape 频率耦合、不可控。改为定时刷进 {@link #queueDepth}
|
||||||
|
* AtomicInteger，把查询频率钉死在固定周期，scrape 时只读内存零 DB——符合「勿每次 scrape 打全表重查」。
|
||||||
|
* 复用既有控制平面计数 {@code countGlobalInflight}（D12 门②背压同一口径），不新增查询。
|
||||||
|
*
|
||||||
|
* <p><b>跨租户</b>：定时线程无租户上下文，须 {@code TenantUtils.executeIgnore} 包裹取全局在飞口径
|
||||||
|
* （与执行器 watchdog 收尸扫描同款）。<b>best-effort</b>：刷新失败只留痕、保留上次值，绝不外抛。
|
||||||
|
*/
|
||||||
|
@Scheduled(fixedRateString = "${huijing.aigc.metrics.queue-depth-refresh-ms:15000}")
|
||||||
|
public void refreshQueueDepth() {
|
||||||
|
try {
|
||||||
|
// 用 lambda(非方法引用)对齐执行器既有 executeIgnore(Callable) 用法,消除 Runnable/Callable 重载歧义。
|
||||||
|
Long inflight = TenantUtils.executeIgnore(() -> aigcTaskMapper.countGlobalInflight());
|
||||||
|
queueDepth.set(inflight == null ? 0 : inflight.intValue());
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.warn("[GenMetrics] 在飞队列深度刷新失败（保留上次值，不阻断）", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
@ -4,6 +4,7 @@ import com.wanxiang.huijing.game.module.aigc.controller.admin.task.vo.DifyCallba
|
|||||||
import com.wanxiang.huijing.game.module.aigc.dal.dataobject.task.AigcTaskDO;
|
import com.wanxiang.huijing.game.module.aigc.dal.dataobject.task.AigcTaskDO;
|
||||||
import com.wanxiang.huijing.game.module.aigc.dal.mysql.task.AigcTaskMapper;
|
import com.wanxiang.huijing.game.module.aigc.dal.mysql.task.AigcTaskMapper;
|
||||||
import com.wanxiang.huijing.game.module.aigc.enums.AigcTaskStatusEnum;
|
import com.wanxiang.huijing.game.module.aigc.enums.AigcTaskStatusEnum;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.framework.metrics.GenMetrics;
|
||||||
import com.wanxiang.huijing.game.module.community.api.CommunityNotifyApi;
|
import com.wanxiang.huijing.game.module.community.api.CommunityNotifyApi;
|
||||||
import com.wanxiang.huijing.game.module.studio.api.SourceProjectApi;
|
import com.wanxiang.huijing.game.module.studio.api.SourceProjectApi;
|
||||||
import com.wanxiang.huijing.game.module.studio.dto.SourceProjectLandReqDTO;
|
import com.wanxiang.huijing.game.module.studio.dto.SourceProjectLandReqDTO;
|
||||||
@ -13,6 +14,7 @@ import com.wanxiang.huijing.framework.common.exception.enums.GlobalErrorCodeCons
|
|||||||
import com.wanxiang.huijing.framework.common.pojo.CommonResult;
|
import com.wanxiang.huijing.framework.common.pojo.CommonResult;
|
||||||
import jakarta.annotation.Resource;
|
import jakarta.annotation.Resource;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
import org.springframework.util.StringUtils;
|
import org.springframework.util.StringUtils;
|
||||||
|
|
||||||
@ -85,6 +87,16 @@ public class DifyCallbackServiceImpl implements DifyCallbackService {
|
|||||||
@Resource
|
@Resource
|
||||||
private SourceProjectApi sourceProjectApi;
|
private SourceProjectApi sourceProjectApi;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 生成业务埋点组件（观测阶段四 · 面五;可选注入）。
|
||||||
|
*
|
||||||
|
* <p><b>可选（required=false）的原因</b>：埋点由开关 {@code huijing.aigc.metrics.enabled} 控制，默认关——关时
|
||||||
|
* {@link GenMetrics} 整个不装配({@code @ConditionalOnProperty}),此处注入为 null。故本类在提交后段调用前必判空
|
||||||
|
* ({@link #recordGenMetricsQuietly}),关时天然旁路、零埋点副作用(旁路铁律)。开时(生产 profile)才在席。
|
||||||
|
*/
|
||||||
|
@Autowired(required = false)
|
||||||
|
private GenMetrics genMetrics;
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Boolean handleCallback(DifyCallbackReqVO reqVO) {
|
public Boolean handleCallback(DifyCallbackReqVO reqVO) {
|
||||||
log.info("[handleCallback] 收到 Dify 回调 traceId={}, status={}, templateId={}",
|
log.info("[handleCallback] 收到 Dify 回调 traceId={}, status={}, templateId={}",
|
||||||
@ -144,6 +156,10 @@ public class DifyCallbackServiceImpl implements DifyCallbackService {
|
|||||||
private void postCommitBestEffort(DifyCallbackReqVO reqVO, SourceLanding landing) {
|
private void postCommitBestEffort(DifyCallbackReqVO reqVO, SourceLanding landing) {
|
||||||
try {
|
try {
|
||||||
AigcTaskDO task = aigcTaskMapper.selectByTraceId(reqVO.getTraceId());
|
AigcTaskDO task = aigcTaskMapper.selectByTraceId(reqVO.getTraceId());
|
||||||
|
// ===== 面五 观测埋点(终态收口):post-commit best-effort,独立兜底、与下方通知/回填解耦 =====
|
||||||
|
// 置于 realSuccess 判定之前,使成功/失败/超时收尸(failed+timeout)三类终态都被计入;
|
||||||
|
// genMetrics 关时为 null → 天然旁路;开时其内部再吞一层异常,双保险确保埋点绝不咬主链、不跳过下方源态回填。
|
||||||
|
recordGenMetricsQuietly(reqVO, task);
|
||||||
boolean realSuccess = task != null
|
boolean realSuccess = task != null
|
||||||
&& Objects.equals(task.getStatus(), AigcTaskStatusEnum.SUCCEEDED.getStatus())
|
&& Objects.equals(task.getStatus(), AigcTaskStatusEnum.SUCCEEDED.getStatus())
|
||||||
&& task.getVersionId() != null;
|
&& task.getVersionId() != null;
|
||||||
@ -165,6 +181,28 @@ public class DifyCallbackServiceImpl implements DifyCallbackService {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 面五 观测埋点静默调用（提交后 best-effort）：把终态计数/耗时/成本/门失败交给 {@link GenMetrics}。
|
||||||
|
*
|
||||||
|
* <p><b>双保险不咬主链</b>：① 开关关时 {@code genMetrics} 为 null → 直接旁路,零副作用;② 开时即便 {@code recordTerminal}
|
||||||
|
* 意外抛错,本方法 catch 吞掉——若不兜,异常会冒进 {@code postCommitBestEffort} 外层 catch、连带跳过其后的通知与
|
||||||
|
* 源态回填(那才是真正咬主链)。故此处必须独立吞异常,不能省。
|
||||||
|
*
|
||||||
|
* @param reqVO 回调入参(取 trace 载荷)
|
||||||
|
* @param task 提交后回查到的任务(取终态 status/failureReason;可能为 null)
|
||||||
|
*/
|
||||||
|
private void recordGenMetricsQuietly(DifyCallbackReqVO reqVO, AigcTaskDO task) {
|
||||||
|
if (genMetrics == null) {
|
||||||
|
return; // 埋点开关关 / 组件未装配 → 零埋点副作用(旁路铁律)
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
genMetrics.recordTerminal(reqVO, task);
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error("[handleCallback] 生成业务埋点失败(best-effort,不影响主链、不跳过源态回填)traceId={}",
|
||||||
|
reqVO.getTraceId(), e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 本次落源结果（U2/B8 §5.2）：持源行 ID + gameId + sourceHash，供「建包成功回填 version_id」「建包失败标孤儿」复用。
|
* 本次落源结果（U2/B8 §5.2）：持源行 ID + gameId + sourceHash，供「建包成功回填 version_id」「建包失败标孤儿」复用。
|
||||||
* landing 为 null（reqVO 无 sourceProject / 落源失败）→ 后续两步均旁路（best-effort，不阻断主链）。
|
* landing 为 null（reqVO 无 sourceProject / 落源失败）→ 后续两步均旁路（best-effort，不阻断主链）。
|
||||||
|
|||||||
@ -0,0 +1,260 @@
|
|||||||
|
package com.wanxiang.huijing.game.module.aigc.framework.metrics;
|
||||||
|
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.controller.admin.task.vo.DifyCallbackReqVO;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.dal.dataobject.task.AigcTaskDO;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.dal.mysql.task.AigcTaskMapper;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.enums.AigcTaskStatusEnum;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.service.callback.DifyCallbackServiceImpl;
|
||||||
|
import com.wanxiang.huijing.game.module.aigc.service.callback.DifyCallbackTxService;
|
||||||
|
import com.wanxiang.huijing.game.module.community.api.CommunityNotifyApi;
|
||||||
|
import io.micrometer.core.instrument.MeterRegistry;
|
||||||
|
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||||
|
import org.springframework.test.util.ReflectionTestUtils;
|
||||||
|
|
||||||
|
import java.util.LinkedHashMap;
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.Mockito.doThrow;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* {@link GenMetrics}（观测阶段四 · 面五 生成业务埋点）单元测试（纯 Mockito + SimpleMeterRegistry，不依赖 DB/Spring 全量上下文）
|
||||||
|
*
|
||||||
|
* <p>覆盖三条硬要求：① 开关开时终态各 status 正确递增 gen_task_total（含超时还原 timed_out）+ 耗时/成本/门失败真落表；
|
||||||
|
* ② 开关关时组件不装配（ApplicationContextRunner 断言 Bean 缺席）；③ 埋点内部抛错不影响回调主流程返回与源态回填。
|
||||||
|
*
|
||||||
|
* @author 绘境AI（观测阶段四 · 面五）
|
||||||
|
*/
|
||||||
|
class GenMetricsTest {
|
||||||
|
|
||||||
|
// ============================== 用例①-a：gen_task_total{status} 按终态递增 ==============================
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 三类终态各自递增对应 status 标签：succeeded / failed(非超时) / timed_out(failed+failureReason=timeout 还原)。
|
||||||
|
* 坐实成功率分母口径 = succeeded + failed + timed_out，且超时据真实 failureReason 还原、非编造。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
void testGenTaskTotal_incrementsByTerminalStatus() {
|
||||||
|
SimpleMeterRegistry registry = new SimpleMeterRegistry();
|
||||||
|
GenMetrics metrics = new GenMetrics(registry, mock(AigcTaskMapper.class));
|
||||||
|
|
||||||
|
metrics.recordTerminal(reqWithTrace(true, 1), taskOf(AigcTaskStatusEnum.SUCCEEDED, null));
|
||||||
|
metrics.recordTerminal(reqWithTrace(false, 3), taskOf(AigcTaskStatusEnum.FAILED, "llm_error"));
|
||||||
|
metrics.recordTerminal(reqWithTrace(false, 3), taskOf(AigcTaskStatusEnum.FAILED, "timeout"));
|
||||||
|
|
||||||
|
assertEquals1(registry.get(GenMetrics.M_TASK).tag("status", "succeeded").counter().count());
|
||||||
|
assertEquals1(registry.get(GenMetrics.M_TASK).tag("status", "failed").counter().count());
|
||||||
|
// 超时:status 落 failed(3)、failureReason=timeout → 还原为 timed_out 标签
|
||||||
|
assertEquals1(registry.get(GenMetrics.M_TASK).tag("status", "timed_out").counter().count());
|
||||||
|
}
|
||||||
|
|
||||||
|
// ============================== 用例①-b：gen_duration_seconds / llm_cost 真落表 ==============================
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 单次终态回调 → gen_duration_seconds 记入一笔（≈trace.wallS 秒）、llm_cost 记入一笔（≈trace.cost.totalRmb）。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
void testDurationAndCost_recordedFromTrace() {
|
||||||
|
SimpleMeterRegistry registry = new SimpleMeterRegistry();
|
||||||
|
GenMetrics metrics = new GenMetrics(registry, mock(AigcTaskMapper.class));
|
||||||
|
|
||||||
|
metrics.recordTerminal(reqWithTrace(true, 1), taskOf(AigcTaskStatusEnum.SUCCEEDED, null));
|
||||||
|
|
||||||
|
// 耗时:count=1，总时长≈wallS(23.4s)
|
||||||
|
assertTrue(registry.get(GenMetrics.M_DURATION).timer().count() == 1L, "gen_duration 应记入一笔");
|
||||||
|
double sec = registry.get(GenMetrics.M_DURATION).timer().totalTime(TimeUnit.SECONDS);
|
||||||
|
assertTrue(Math.abs(sec - 23.4) < 0.1, "gen_duration 总时长应≈wallS=23.4s，实际=" + sec);
|
||||||
|
|
||||||
|
// 成本:count=1，总额≈totalRmb(0.031)
|
||||||
|
assertTrue(registry.get(GenMetrics.M_LLM_COST).summary().count() == 1L, "llm_cost 应记入一笔");
|
||||||
|
double rmb = registry.get(GenMetrics.M_LLM_COST).summary().totalAmount();
|
||||||
|
assertTrue(Math.abs(rmb - 0.031) < 1e-6, "llm_cost 总额应≈totalRmb=0.031，实际=" + rmb);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ============================== 用例①-c：gen_gate_fail_total{gate} 逐门失败 ==============================
|
||||||
|
|
||||||
|
/**
|
||||||
|
* guards = {C_frame:true(过), H_progress:{pass:false}(嵌套·未过), E_live:false(裸·未过)}
|
||||||
|
* → 仅两个未过门各 +1；过门不计;嵌套 H_progress 取 .pass 判定(复用 ReadinessScorer 口径)。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
void testGenGateFail_perFailingGate() {
|
||||||
|
SimpleMeterRegistry registry = new SimpleMeterRegistry();
|
||||||
|
GenMetrics metrics = new GenMetrics(registry, mock(AigcTaskMapper.class));
|
||||||
|
|
||||||
|
DifyCallbackReqVO reqVO = new DifyCallbackReqVO();
|
||||||
|
reqVO.setTraceId("t-gate");
|
||||||
|
Map<String, Object> guards = new LinkedHashMap<>();
|
||||||
|
guards.put("C_frame", true); // 过 → 不计
|
||||||
|
guards.put("H_progress", Map.of("pass", false)); // 嵌套未过 → +1
|
||||||
|
guards.put("E_live", false); // 裸未过 → +1
|
||||||
|
Map<String, Object> verdict = new LinkedHashMap<>();
|
||||||
|
verdict.put("guards", guards);
|
||||||
|
Map<String, Object> trace = new LinkedHashMap<>();
|
||||||
|
trace.put("sevenGateVerdict", verdict);
|
||||||
|
reqVO.setTrace(trace);
|
||||||
|
|
||||||
|
metrics.recordTerminal(reqVO, taskOf(AigcTaskStatusEnum.FAILED, "llm_error"));
|
||||||
|
|
||||||
|
assertEquals1(registry.get(GenMetrics.M_GATE_FAIL).tag("gate", "H_progress").counter().count());
|
||||||
|
assertEquals1(registry.get(GenMetrics.M_GATE_FAIL).tag("gate", "E_live").counter().count());
|
||||||
|
// 过门不产计数器
|
||||||
|
assertNull(registry.find(GenMetrics.M_GATE_FAIL).tag("gate", "C_frame").counter(),
|
||||||
|
"过门 C_frame 不应产失败计数");
|
||||||
|
}
|
||||||
|
|
||||||
|
// ============================== 用例①-d：gen_queue_depth 定时刷 → Gauge 读内存 ==============================
|
||||||
|
|
||||||
|
/**
|
||||||
|
* refreshQueueDepth 读全局在飞计数刷 AtomicInteger，Gauge 暴露该值(scrape 时零 DB)。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
void testQueueDepthGauge_refreshedFromInflightCount() {
|
||||||
|
SimpleMeterRegistry registry = new SimpleMeterRegistry();
|
||||||
|
AigcTaskMapper mapper = mock(AigcTaskMapper.class);
|
||||||
|
when(mapper.countGlobalInflight()).thenReturn(7L);
|
||||||
|
GenMetrics metrics = new GenMetrics(registry, mapper);
|
||||||
|
|
||||||
|
// 刷新前默认 0
|
||||||
|
assertTrue(registry.get(GenMetrics.M_QUEUE_DEPTH).gauge().value() == 0.0, "初始应为 0");
|
||||||
|
metrics.refreshQueueDepth();
|
||||||
|
assertTrue(registry.get(GenMetrics.M_QUEUE_DEPTH).gauge().value() == 7.0, "刷新后应为在飞计数 7");
|
||||||
|
}
|
||||||
|
|
||||||
|
// ============================== 用例①-e：脏 trace 不抛、不产脏计数(埋点内部健壮性) ==============================
|
||||||
|
|
||||||
|
/**
|
||||||
|
* trace 字段类型不符(wallS 非数、cost 非 Map、guards 缺失) → recordTerminal 不抛、不误记，仅 status 计数正常。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
void testMalformedTrace_swallowedNoBogusMetric() {
|
||||||
|
SimpleMeterRegistry registry = new SimpleMeterRegistry();
|
||||||
|
GenMetrics metrics = new GenMetrics(registry, mock(AigcTaskMapper.class));
|
||||||
|
|
||||||
|
DifyCallbackReqVO reqVO = new DifyCallbackReqVO();
|
||||||
|
reqVO.setTraceId("t-bad");
|
||||||
|
Map<String, Object> trace = new LinkedHashMap<>();
|
||||||
|
trace.put("wallS", "not-a-number");
|
||||||
|
trace.put("cost", "not-a-map");
|
||||||
|
reqVO.setTrace(trace);
|
||||||
|
|
||||||
|
assertDoesNotThrow(() -> metrics.recordTerminal(reqVO, taskOf(AigcTaskStatusEnum.SUCCEEDED, null)));
|
||||||
|
// status 仍正常计;耗时/成本 meter 虽在构造期已注册(供复用),但脏数据一笔都不应记入(count=0)
|
||||||
|
assertEquals1(registry.get(GenMetrics.M_TASK).tag("status", "succeeded").counter().count());
|
||||||
|
assertTrue(registry.get(GenMetrics.M_DURATION).timer().count() == 0L, "非数 wallS 不应记入 gen_duration");
|
||||||
|
assertTrue(registry.get(GenMetrics.M_LLM_COST).summary().count() == 0L, "非 Map cost 不应记入 llm_cost");
|
||||||
|
}
|
||||||
|
|
||||||
|
// ============================== 用例②：开关控制装配(关时组件不装配) ==============================
|
||||||
|
|
||||||
|
/**
|
||||||
|
* ApplicationContextRunner 断言 @ConditionalOnProperty 行为：
|
||||||
|
* 缺省/显式 false → GenMetrics Bean 缺席(关时零埋点副作用);true → 装配。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
void testConditionalAssembly_offByDefault_onWhenEnabled() {
|
||||||
|
ApplicationContextRunner runner = new ApplicationContextRunner()
|
||||||
|
.withBean(MeterRegistry.class, SimpleMeterRegistry::new)
|
||||||
|
.withBean(AigcTaskMapper.class, () -> mock(AigcTaskMapper.class))
|
||||||
|
.withUserConfiguration(GenMetrics.class);
|
||||||
|
|
||||||
|
// 缺省(未设开关)→ 不装配
|
||||||
|
runner.run(context -> assertThat(context).doesNotHaveBean(GenMetrics.class));
|
||||||
|
// 显式 false → 不装配
|
||||||
|
runner.withPropertyValues("huijing.aigc.metrics.enabled=false")
|
||||||
|
.run(context -> assertThat(context).doesNotHaveBean(GenMetrics.class));
|
||||||
|
// 显式 true → 装配
|
||||||
|
runner.withPropertyValues("huijing.aigc.metrics.enabled=true")
|
||||||
|
.run(context -> assertThat(context).hasSingleBean(GenMetrics.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
// ============================== 用例③：埋点抛错不咬回调主链(红线) ==============================
|
||||||
|
|
||||||
|
/**
|
||||||
|
* genMetrics.recordTerminal 抛错 → 回调 handleCallback 仍 return TRUE，且提交后的源态通知照常发出
|
||||||
|
* (证明埋点异常被独立吞、不冒进外层 catch 跳过后续 notify/回填)——旁路铁律硬红线。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
void testGenMetricsThrows_doesNotBlockCallback() {
|
||||||
|
DifyCallbackTxService txService = mock(DifyCallbackTxService.class);
|
||||||
|
when(txService.handleCallbackTx(any(DifyCallbackReqVO.class))).thenReturn(Boolean.TRUE);
|
||||||
|
DifyCallbackServiceImpl callbackService = new DifyCallbackServiceImpl(txService);
|
||||||
|
|
||||||
|
AigcTaskMapper mapper = mock(AigcTaskMapper.class);
|
||||||
|
AigcTaskDO succeeded = taskOf(AigcTaskStatusEnum.SUCCEEDED, null);
|
||||||
|
succeeded.setCreatorUserId(1000L);
|
||||||
|
succeeded.setGameId(5L);
|
||||||
|
succeeded.setVersionId(2048L); // realSuccess=true → 提交后 notify 会触发
|
||||||
|
when(mapper.selectByTraceId("t-red")).thenReturn(succeeded);
|
||||||
|
CommunityNotifyApi notifyApi = mock(CommunityNotifyApi.class);
|
||||||
|
|
||||||
|
// 抛错的埋点组件
|
||||||
|
GenMetrics throwing = mock(GenMetrics.class);
|
||||||
|
doThrow(new RuntimeException("埋点炸了")).when(throwing).recordTerminal(any(), any());
|
||||||
|
|
||||||
|
ReflectionTestUtils.setField(callbackService, "aigcTaskMapper", mapper);
|
||||||
|
ReflectionTestUtils.setField(callbackService, "communityNotifyApi", notifyApi);
|
||||||
|
ReflectionTestUtils.setField(callbackService, "genMetrics", throwing);
|
||||||
|
|
||||||
|
DifyCallbackReqVO reqVO = reqWithTrace(true, 1);
|
||||||
|
reqVO.setTraceId("t-red"); // 无 sourceProject → 落源整段旁路,不需 mock sourceProjectApi
|
||||||
|
|
||||||
|
// 红线:埋点抛错被吞,主链照常 return TRUE
|
||||||
|
assertTrue(callbackService.handleCallback(reqVO), "埋点抛错不应影响回调主流程返回");
|
||||||
|
// 埋点确实被调(且抛了,但被独立吞)
|
||||||
|
verify(throwing).recordTerminal(any(), any());
|
||||||
|
// 提交后源态通知照常发出(证明埋点异常没有跳过后续 best-effort 步骤)
|
||||||
|
verify(notifyApi).notifyGenerateDone(1000L, 5L, 2048L);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ============================== 测试夹具 ==============================
|
||||||
|
|
||||||
|
/** 断计数器为 1（micrometer count 返回 double）。 */
|
||||||
|
private static void assertEquals1(double actual) {
|
||||||
|
assertTrue(actual == 1.0, "计数应为 1，实际=" + actual);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 构造指定终态 + failureReason 的任务 DO。 */
|
||||||
|
private static AigcTaskDO taskOf(AigcTaskStatusEnum status, String failureReason) {
|
||||||
|
AigcTaskDO task = new AigcTaskDO();
|
||||||
|
task.setId(77L);
|
||||||
|
task.setTraceId("t-x");
|
||||||
|
task.setStatus(status.getStatus());
|
||||||
|
task.setFailureReason(failureReason);
|
||||||
|
return task;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 构造带 9d trace（wallS/cost/sevenGateVerdict.guards）的回调入参。 */
|
||||||
|
private static DifyCallbackReqVO reqWithTrace(boolean pass, int repairs) {
|
||||||
|
DifyCallbackReqVO reqVO = new DifyCallbackReqVO();
|
||||||
|
reqVO.setTraceId("t-x");
|
||||||
|
reqVO.setStatus(pass ? "succeeded" : "failed");
|
||||||
|
Map<String, Object> trace = new LinkedHashMap<>();
|
||||||
|
trace.put("pass", pass);
|
||||||
|
trace.put("repairs", repairs);
|
||||||
|
trace.put("wallS", 23.4);
|
||||||
|
Map<String, Object> guards = new LinkedHashMap<>();
|
||||||
|
guards.put("C_frame", true);
|
||||||
|
guards.put("H_progress", Map.of("pass", pass));
|
||||||
|
Map<String, Object> verdict = new LinkedHashMap<>();
|
||||||
|
verdict.put("pass", pass);
|
||||||
|
verdict.put("guards", guards);
|
||||||
|
trace.put("sevenGateVerdict", verdict);
|
||||||
|
Map<String, Object> cost = new LinkedHashMap<>();
|
||||||
|
cost.put("totalRmb", 0.031);
|
||||||
|
trace.put("cost", cost);
|
||||||
|
reqVO.setTrace(trace);
|
||||||
|
return reqVO;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
Loading…
x
Reference in New Issue
Block a user