夜雨聆风🚀 Day14: 极限异常演练 —— 验证系统韧性与错误告警生成
今日目标： 在企业级架构中，真正的考验不在于系统正常运行时有多快，而在于系统出错时死得有多优雅。 今天，我们要实施一套“错误状态注入（Error State Injection）”机制。如果 API 挂了、网络断开，或者大模型（LLM）产生了 JSON 结构幻觉（Schema Hallucination），系统绝不能吞掉错误或假死。相反，它必须捕获致命崩溃，并将一条明确的 event_type="PEAK_Error" 日志写入 main 索引，作为发送给 SOC 大屏的 SOS 信号！ 此外，我们还将解决一个真实的由 LLM 结构幻觉 引起的崩溃问题（例如：模型返回了一个字符串而不是嵌套字典），并实施防弹级别的类型检查防御。
💻 终极实战：Day 14 鲁棒性增强版代码基线
请打开 Add-on Builder 的 Define & Test 编辑器，用以下版本替换您现有的代码。
import osimport sysimport timeimport datetimeimport jsonimport uuidimport requestsimport splunklib.client as clientimport splunklib.results as results# ==========================================# HELPER 1: Execute AI Generated SPL# ==========================================def execute_ai_spl(helper, service, spl_query):"""Execute SPL generated by AI and return the raw result data."""spl_query = spl_query.strip()if not spl_query.startswith("search") and not spl_query.startswith("|"):spl_query = "search " + spl_querykwargs_oneshot = {"output_mode": "json"}helper.log_info(f"[Agentic Engine] Executing SPL: {spl_query}")try:search_results = service.jobs.oneshot(spl_query, **kwargs_oneshot)reader = results.JSONResultsReader(search_results)result_data = [res for res in reader if isinstance(res, dict)]helper.log_info(f"[Agentic Engine] SUCCESS: Found {len(result_data)} events.")return result_dataexcept Exception as e:helper.log_error(f"[Agentic Engine] FAILED execution: {str(e)}")return []# ==========================================# HELPER 2: Fetch Real Logs (M-ATH Concept)# ==========================================def fetch_rare_logs(helper, service, target_index):"""Fetch the most recent rare/anomalous logs from the target index."""helper.log_info("Fetching real rare logs for analysis...")spl = f"search index={target_index} | head 5 | table _raw"try:results_data = execute_ai_spl(helper, service, spl)if not results_data:return Noneraw_logs = [item.get("_raw", "") for item in results_data if "_raw" in item]payload = "\n".join(raw_logs)# Context Distillation (Payload Truncation)MAX_CHARS = 6000if len(payload) > MAX_CHARS:helper.log_info(f"Payload too large ({len(payload)} chars). Truncating to {MAX_CHARS}...")payload = payload[:MAX_CHARS] + "\n\n...[TRUNCATED DUE TO CONTEXT LIMITS. ANALYZE AVAILABLE DATA ONLY.]..."return payloadexcept Exception as e:helper.log_error(f"Failed to fetch rare logs: {str(e)}")return None# =========================================================================# Universal Token Extractor (FinOps Cost Tracking)# =========================================================================def extract_token_usage(helper, response_json, response_headers):"""Robustly extract token usage across different LLM providers and API gateways.Ensures FinOps tracking never crashes the main thread."""try:if "usage" in response_json:usage = response_json["usage"]if "total_tokens" in usage:return int(usage["total_tokens"])elif "prompt_tokens" in usage and "completion_tokens" in usage:return int(usage["prompt_tokens"]) + int(usage["completion_tokens"])elif "input_tokens" in usage and "output_tokens" in usage:return int(usage["input_tokens"]) + int(usage["output_tokens"])header_keys = [k.lower() for k in response_headers.keys()]for key in header_keys:if "token-usage" in key or "x-ratelimit-usage" in key:return int(response_headers.get(key, 0))except Exception as e:helper.log_error(f"[FinOps Warning] Failed to parse token usage correctly: {str(e)}")return 0# ==========================================# HELPER 3: The LLM API Connector# ==========================================def call_llm_api(helper, api_key, base_url, model, system_prompt, user_prompt, max_tokens):"""Establish real HTTP connection to the LLM API and return the JSON response."""headers = {"Authorization": f"Bearer {api_key}","Content-Type": "application/json"}payload = {"model": model,"messages": [{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}],"response_format": {"type": "json_object"},"max_tokens": max_tokens}endpoint = base_url if base_url.endswith("/chat/completions") else f"{base_url.rstrip('/')}/chat/completions"try:helper.log_info(f"Initiating network request to LLM API: {endpoint} (Max Tokens: {max_tokens})")response = requests.post(endpoint, headers=headers, json=payload, timeout=120)# Raises HTTPError for bad responses (4xx or 5xx), triggering the global exception handlerresponse.raise_for_status()response_json = response.json()llm_content = response_json["choices"][0]["message"]["content"]total_tokens = extract_token_usage(helper, response_json, response.headers)helper.log_info(f"API Call Success. FinOps Tracked: {total_tokens} tokens consumed.")return llm_content, total_tokensexcept requests.exceptions.RequestException as e:helper.log_error(f"Network error during API call: {str(e)}")raise# ==========================================# MAIN WORKFLOW: The Autonomous Agent# ==========================================def collect_events(helper, ew):"""The Ultimate Live Workflow.Features: API Integration, Epoch Time, Anti-Hallucination, Truncation, FinOps, and Chaos Resilience."""helper.log_info("PEAK AI Hunter: LIVE MODE INITIALIZED.")cycle_start_time = time.time()hunt_session_id = str(uuid.uuid4())try:session_key = getattr(helper, 'session_key', None) or getattr(helper._input_definition, 'metadata', {}).get('session_key')if not session_key:raise ValueError("Failed to acquire session_key from Splunk core.")service = client.Service(token=session_key)api_key = helper.get_global_setting("api_key")base_url = helper.get_global_setting("base_url")model_name = helper.get_global_setting("model_name")target_index = helper.get_output_index() or "main"if not api_key or not base_url:raise ValueError("API Key or Base URL is missing in Global Settings.")# ==========================================# PHASE 1: PREPARE (Blueprint Generation)# ==========================================rare_logs_payload = fetch_rare_logs(helper, service, target_index)if not rare_logs_payload:helper.log_info("No anomalous logs found to analyze. Terminating cycle early.")return# Added strict constraint for 'ABLE' to prevent Schema Hallucinationsys_prompt_prepare = "You are a Senior Threat Hunter. You MUST reply in JSON format. Be extremely concise. No pleasantries. Schema: 'analysis' (string), 'hypotheses' (array). Each hypothesis MUST have 'ABLE' (must be a nested JSON object with keys: Actor, Behavior, Location, Evidence), 'spl_round_1_validation', and 'spl_round_2_drilldown'."usr_prompt_prepare = f"Analyze these logs:\n{rare_logs_payload}\n\nGenerate exactly 2 hypotheses. CRITICAL: For SPL, strictly start with 'search index={{target_index}}'. Output ONLY JSON."helper.log_info("Triggering LLM for Prepare Phase...")blueprint_text, prep_tokens = call_llm_api(helper, api_key, base_url, model_name, sys_prompt_prepare, usr_prompt_prepare, max_tokens=1500)ai_hunting_plan = json.loads(blueprint_text.strip())hypotheses = ai_hunting_plan.get("hypotheses", [])ew.write_event(helper.new_event(source=helper.get_input_type(), index=target_index, sourcetype="_json",time=time.time(),data=json.dumps({"session_id": hunt_session_id, "event_type": "PEAK_Plan", "timestamp": round(time.time(), 3), "content": ai_hunting_plan}, ensure_ascii=False)))# ==========================================# PHASE 2: EXECUTE (Autonomous Query Loop)# ==========================================all_hunt_evidence = []for i, hyp in enumerate(hypotheses):hyp_start = time.time()spl_r1 = hyp.get("spl_round_1_validation", "").replace("{target_index}", target_index)spl_r2 = hyp.get("spl_round_2_drilldown", "").replace("{target_index}", target_index)r1_hits = len(execute_ai_spl(helper, service, spl_r1))r2_hits = len(execute_ai_spl(helper, service, spl_r2))# =========================================================================# [DAY 14 HOTFIX]: LLM Schema Hallucination Defense (Defensive Programming)# Safely extract behavior regardless of whether LLM returned a Dict or a flattened String.# Prevents: AttributeError: 'str' object has no attribute 'get'# =========================================================================able_data = hyp.get('ABLE', {})if isinstance(able_data, dict):behavior_text = able_data.get('Behavior', 'Unknown')else:# If the LLM hallucinated and flattened ABLE into a single stringbehavior_text = str(able_data)# =========================================================================all_hunt_evidence.append({"hypothesis_id": hyp.get("hypothesis_id", i+1),"threat_behavior": behavior_text, # <--- Safe ingestion"round_1_hit_count": r1_hits,"round_2_hit_count": r2_hits,"execution_duration_sec": round(time.time() - hyp_start, 2)})ew.write_event(helper.new_event(source=helper.get_input_type(), index=target_index, sourcetype="_json",time=time.time(),data=json.dumps({"session_id": hunt_session_id, "event_type": "PEAK_Evidence", "timestamp": round(time.time(), 3), "content": all_hunt_evidence}, ensure_ascii=False)))# ==========================================# PHASE 3: ACT (Final Report Generation)# ==========================================sys_prompt_act = "You are a Security Director. Output ONLY valid JSON. Keep summaries under 30 words. Keys: 'executive_summary', 'threat_qualification', 'risk_score', 'recommended_alert_spl'."usr_prompt_act = f"Here is the execution evidence:\n{json.dumps(all_hunt_evidence)}\n\nBased on these hits, qualify the threat, assign a score, and write alert SPL. Reply in JSON."helper.log_info("Triggering LLM for Act Phase...")report_text, act_tokens = call_llm_api(helper, api_key, base_url, model_name, sys_prompt_act, usr_prompt_act, max_tokens=800)try:final_report = json.loads(report_text.strip())except json.JSONDecodeError as e:helper.log_error("JSON Truncation in Act Phase. Engaging fallback.")final_report = {"executive_summary": "LLM output truncated.", "risk_score": -1, "raw": report_text}ew.write_event(helper.new_event(source=helper.get_input_type(), index=target_index, sourcetype="_json",time=time.time(),data=json.dumps({"session_id": hunt_session_id, "event_type": "PEAK_Final_Report", "timestamp": round(time.time(), 3), "total_tokens_used": prep_tokens + act_tokens, "content": final_report}, ensure_ascii=False)))helper.log_info(f"LIVE CYCLE COMPLETE. Time: {round(time.time() - cycle_start_time, 2)}s. Session ID: {hunt_session_id}")except Exception as e:# =========================================================================# [DAY 14 NEW]: Enterprise-Grade Graceful Degradation & Error Alerting# When a fatal error occurs, we write an explicit Error Event to the frontend# so the SOC dashboard goes RED and operators are notified immediately.# =========================================================================error_msg = str(e)helper.log_error(f"FATAL Pipeline Crash: {error_msg}")try:fallback_index = helper.get_output_index() or "main"ew.write_event(helper.new_event(source=helper.get_input_type(), index=fallback_index, sourcetype="_json",time=time.time(),data=json.dumps({"session_id": hunt_session_id,"event_type": "PEAK_Error","timestamp": round(time.time(), 3),"error_message": error_msg,"agent_status": "CRITICAL_FAILURE"}, ensure_ascii=False)))helper.log_info("[Chaos Engineering] Sent Error_State alert to main index successfully.")except Exception as write_err:helper.log_error(f"Secondary Crash: Could not write PEAK_Error event. {str(write_err)}")
💣 混沌工程演练：破坏性验证 (Destructive Validation)
保存上面的代码后，我们将故意破坏环境，以确保我们的系统能够安全失败并触发 PEAK_Error 告警。
💥 演练 1：结构幻觉 (代码级防御)
●根本原因 (The Root Cause)：像 GPT-4 或 Qwen 这样的 LLM 有时会遭遇“结构幻觉 (Schema Hallucination)”，返回一个扁平化的字符串而不是嵌套字典（例如，返回 "ABLE": "Some text..." 而不是 "ABLE": {"Behavior": "..."}）。当 Python 试图对字符串调用 .get('Behavior') 时，就会抛出 AttributeError: 'str' object has no attribute 'get'。
●验证逻辑 (Validation)：[DAY 14 HOTFIX] 使用 isinstance() 嗅探数据类型。如果 LLM 发生暴走并返回了一个字符串，我们的代码会通过 behavior_text = str(able_data) 优雅地接收它，并继续运行而不会崩溃。
💥 演练 2：API 密钥泄露与吊销 (网络级防御)
●破坏动作 (Destructive Action)：前往 AOB 的 Configuration 页面，故意破坏你的 API Key（例如删除几个字母），然后点击 Save。
●执行 (Execution)：在代码编辑器中点击 Test。
●预期现象 (Expected Phenomenon)：底层的 requests.raise_for_status() 将捕获到 401 Unauthorized。[DAY 14 NEW] 全局异常处理器将接管该异常，防止发生静默崩溃，并写入一条 CRITICAL_FAILURE 日志。
💥 演练 3：DNS 劫持 / 网络黑洞
●破坏动作 (Destructive Action)：将 Base URL 更改为一个无法访问的 IP（例如 https://10.255.255.1）。
●预期现象 (Expected Phenomenon)：在 120 秒超时后，系统抛出 ConnectTimeout。全局处理器将其捕获并向前端发出告警。
🚨 验收战果：在前端查看您的“灾难报告”
完成演练后，请将您的 API Key 和 Base URL 恢复为正确的值。 然后，打开 Splunk Search 界面并执行此系统健康监控 SPL（时间范围：Last 15 minutes）：
index=main sourcetype="_json" event_type="PEAK_Error"| table timestamp, session_id, agent_status, error_message| eval timestamp=strftime(timestamp, "%Y-%m-%d %H:%M:%S")| rename timestamp as "Crash Time", session_id as "Failed Session ID", agent_status as "Status", error_message as "Root Cause Analysis"
🎯 您的验收时刻：您将看到一张极其专业的系统熔断报表 (System Trip Report)！它清晰地记录了您所模拟的 AttributeError、401 认证失败或网络超时。 这正是具备韧性系统的标志：它不惧怕失败，它只惧怕默默无闻地死去。 有了这种告警机制和类型安全补丁的加持，您的引擎现在几乎是坚不可摧的！
👇 全套教程多平台同步更新 👇
https://blog.csdn.net/thewindrider/category_13144991.html
https://gitcode.com/Chang_feng_Po/Splunk-AI-PEAK-Tutorial
