package analyzer import "fmt" // Analyze runs all heuristic rules and returns the triggered findings. func Analyze(in Input, t Thresholds) []Finding { var out []Finding out = append(out, checkDataSkew(in, t)...) out = append(out, checkGCPressure(in, t)...) out = append(out, checkBottleneck(in, t)...) return out } func checkDataSkew(in Input, t Thresholds) []Finding { var out []Finding for stage, m := range in.StageMetrics { if m.MinPartitionBytes <= 0 || m.MaxPartitionBytes <= 0 { continue } ratio := float64(m.MaxPartitionBytes) / float64(m.MinPartitionBytes) if ratio > t.DataSkewRatio { sev := SeverityWarning if ratio > t.DataSkewRatio*2 { sev = SeverityCritical } out = append(out, Finding{ Rule: "data_skew", Severity: sev, Stage: stage, Evidence: map[string]any{ "max_partition_bytes": m.MaxPartitionBytes, "min_partition_bytes": m.MinPartitionBytes, "ratio": ratio, "threshold": t.DataSkewRatio, }, Message: fmt.Sprintf("stage %q 数据倾斜: max/min = %.1fx (阈值 %.1fx)", stage, ratio, t.DataSkewRatio), }) } } return out } func checkGCPressure(in Input, t Thresholds) []Finding { var out []Finding var totalGC, totalCPU int64 var worstExec string var worstRatio float64 for id, m := range in.ExecutorMetrics { if m.CPUTimeMS <= 0 { continue } r := float64(m.GCTimeMS) / float64(m.CPUTimeMS) totalGC += m.GCTimeMS totalCPU += m.CPUTimeMS if r > worstRatio { worstRatio = r worstExec = id } } if totalCPU > 0 { aggRatio := float64(totalGC) / float64(totalCPU) if aggRatio > t.GCPressureRatio { sev := SeverityWarning if aggRatio > t.GCPressureRatio*2 { sev = SeverityCritical } out = append(out, Finding{ Rule: "gc_pressure", Severity: sev, Evidence: map[string]any{ "aggregate_gc_ratio": aggRatio, "worst_executor": worstExec, "worst_ratio": worstRatio, "threshold": t.GCPressureRatio, }, Message: fmt.Sprintf("GC 压力过高: 聚合 GC/CPU 比率 = %.1f%% (阈值 %.1f%%, 最差 executor %s = %.1f%%)", aggRatio*100, t.GCPressureRatio*100, worstExec, worstRatio*100), }) } } return out } func checkBottleneck(in Input, t Thresholds) []Finding { var out []Finding thresholdBytes := int64(t.BottleneckShuffleGB * (1 << 30)) for stage, m := range in.StageMetrics { total := m.ShuffleReadBytes + m.ShuffleWriteBytes if total < 0 { continue // overflow guard } if total > thresholdBytes { sev := SeverityInfo if total > thresholdBytes*2 { sev = SeverityWarning } if total > thresholdBytes*5 { sev = SeverityCritical } out = append(out, Finding{ Rule: "bottleneck", Severity: sev, Stage: stage, Evidence: map[string]any{ "shuffle_read_bytes": m.ShuffleReadBytes, "shuffle_write_bytes": m.ShuffleWriteBytes, "shuffle_total_gb": float64(total) / (1 << 30), "threshold_gb": t.BottleneckShuffleGB, }, Message: fmt.Sprintf("stage %q 是潜在瓶颈: shuffle 总 %.1f GB (阈值 %.1f GB)", stage, float64(total)/(1<<30), t.BottleneckShuffleGB), }) } } return out }