夜雨聆风🚀 Day 15: 底层大一统 —— 代码 Review 与跨阶段数据校验
今日目标：
1. 重塑日志引擎：梳理 Python 引擎中的杂乱日志。将里程碑事件设为 INFO，将详细变量/Payload 设为 DEBUG（以便平时静默，排障时开启），将崩溃设为 ERROR。
2. 状态机事务校验：使用 transaction 命令，将原本散落在时间轴上的 PEAK_Plan、PEAK_Evidence、PEAK_Final_Report（或 PEAK_Error）重新“缝合”成一个完整的业务事务，验证系统的状态机流转。
💻 终极实战：后端引擎封版代码 (Production-Ready Release)
这是我们整个 Add-on Builder 的最终版核心代码。请打开 Add-on Builder 的 Define & Test 编辑器，用这套封版代码覆盖原有代码，点击 Save。
import os
import sys
import time
import datetime
import json
import uuid
import requests
import splunklib.client as client
import splunklib.results as results
# ==========================================
# HELPER 1: Execute AI Generated SPL
# ==========================================
def execute_ai_spl(helper, service, spl_query):
"""
ExecuteSPL generated by AI and return the raw result data.
"""
spl_query= spl_query.strip()
#Force the 'search' prefix to prevent syntax errors
ifnot spl_query.startswith("search") and not spl_query.startswith("|"):
spl_query= "search " + spl_query
kwargs_oneshot= {"output_mode": "json"}
helper.log_debug(f"[AgenticEngine] 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"[AgenticEngine] SUCCESS: Found {len(result_data)} events.")
returnresult_data
exceptException as e:
helper.log_error(f"[AgenticEngine] FAILED execution: {str(e)}")
return[]
# ==========================================
# HELPER 2: Fetch Real Logs (M-ATH Concept)
# ==========================================
def fetch_rare_logs(helper, service, target_index):
"""
Fetchthe most recent rare/anomalous logs from the target index.
Includescontext distillation to prevent context window overflow.
"""
helper.log_info("Fetchingreal rare logs for analysis...")
spl= f"search index={target_index} | head 5 | table _raw"
try:
results_data= execute_ai_spl(helper, service, spl)
ifnot results_data:
helper.log_debug("Nologs returned from target index.")
returnNone
raw_logs= [item.get("_raw", "") for item in results_data if "_raw" in item]
payload= "\n".join(raw_logs)
#Context Distillation (Payload Truncation)
MAX_CHARS= 6000
iflen(payload) > MAX_CHARS:
helper.log_info(f"Payloadtoo large ({len(payload)} chars). Truncating to {MAX_CHARS}...")
payload= payload[:MAX_CHARS] + "\n\n...[TRUNCATED DUE TO CONTEXT LIMITS. ANALYZE AVAILABLE DATA ONLY.]..."
returnpayload
exceptException as e:
helper.log_error(f"Failedto fetch rare logs: {str(e)}")
returnNone
# =========================================================================
# HELPER 3: Universal Token Extractor (FinOps Cost Tracking)
# =========================================================================
def extract_token_usage(helper, response_json, response_headers):
"""
Robustlyextract token usage across different LLM providers and API gateways.
EnsuresFinOps tracking never crashes the main thread.
"""
try:
if"usage" in response_json:
usage= response_json["usage"]
if"total_tokens" in usage:
returnint(usage["total_tokens"])
elif"prompt_tokens" in usage and "completion_tokens" in usage:
returnint(usage["prompt_tokens"]) + int(usage["completion_tokens"])
elif"input_tokens" in usage and "output_tokens" in usage:
returnint(usage["input_tokens"]) + int(usage["output_tokens"])
header_keys= [k.lower() for k in response_headers.keys()]
forkey in header_keys:
if"token-usage" in key or "x-ratelimit-usage" in key:
returnint(response_headers.get(key, 0))
exceptException as e:
helper.log_error(f"[FinOpsWarning] Failed to parse token usage correctly: {str(e)}")
return0
# ==========================================
# HELPER 4: The LLM API Connector
# ==========================================
def call_llm_api(helper, api_key, base_url, model, system_prompt, user_prompt, max_tokens):
"""
Establishreal HTTP connection to the LLM API.
Includeshardware-level token circuit breakers and graceful timeouts.
"""
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"Initiatingnetwork request to LLM API: {endpoint} (Max Tokens: {max_tokens})")
helper.log_debug(f"LLMPayload size: {len(user_prompt)} chars.")
response= requests.post(endpoint, headers=headers, json=payload, timeout=120)
response.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"APICall Success. FinOps Tracked: {total_tokens} tokens consumed.")
returnllm_content, total_tokens
exceptrequests.exceptions.RequestException as e:
helper.log_error(f"Networkerror during API call: {str(e)}")
raise
# ==========================================
# MAIN WORKFLOW: The Autonomous Agent
# ==========================================
def collect_events(helper, ew):
"""
TheUltimate Live Workflow (Production Release).
Features:Dynamic AI queries, Anti-Hallucination, Truncation, FinOps, and Chaos Resilience.
"""
helper.log_info("PEAKAI Hunter: LIVE MODE INITIALIZED.")
cycle_start_time= time.time()
hunt_session_id= str(uuid.uuid4())
helper.log_debug(f"Generatednew Session ID: {hunt_session_id}")
try:
#Acquire Splunk Service Session
session_key= getattr(helper, 'session_key', None) or getattr(helper._input_definition, 'metadata', {}).get('session_key')
ifnot session_key:
raiseValueError("Failed to acquire session_key from Splunk core.")
service= client.Service(token=session_key)
#Acquire Global Setup Configurations
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"
ifnot api_key or not base_url:
raiseValueError("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)
ifnot rare_logs_payload:
helper.log_info("Noanomalous logs found to analyze. Terminating cycle early.")
return
sys_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("TriggeringLLM 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)
))
helper.log_debug("Successfullywrote PEAK_Plan to Splunk.")
#==========================================
#PHASE 2: EXECUTE (Autonomous Query Loop)
#==========================================
all_hunt_evidence= []
fori, 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))
#Defensive Programming: Safeguard against LLM Schema Hallucinations
able_data= hyp.get('ABLE', {})
ifisinstance(able_data, dict):
behavior_text= able_data.get('Behavior', 'Unknown')
else:
behavior_text= str(able_data)
all_hunt_evidence.append({
"hypothesis_id":hyp.get("hypothesis_id", i+1),
"threat_behavior":behavior_text,
"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)
))
helper.log_debug("Successfullywrote PEAK_Evidence to Splunk.")
#==========================================
#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("TriggeringLLM 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())
exceptjson.JSONDecodeError as e:
helper.log_error("JSONTruncation 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_debug("Successfullywrote PEAK_Final_Report to Splunk.")
helper.log_info(f"LIVECYCLE COMPLETE. Time: {round(time.time() - cycle_start_time, 2)}s. Session ID: {hunt_session_id}")
exceptException as e:
#Enterprise-Grade Graceful Degradation & Error Alerting
error_msg= str(e)
helper.log_error(f"FATALPipeline 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("SentError_State alert to main index successfully.")
exceptException as write_err:
helper.log_error(f"SecondaryCrash: Could not write PEAK_Error event. {str(write_err)}")
🔍 Transaction 跨阶段拼接测试
代码写完且执行过几次 Test（或者等待后台自动调度运行几次）之后，我们要离开后台代码区，来到前端大舞台：Splunk 的 Search 界面。
由于我们的数据写入是“按阶段”拆分的（先写 Plan，再写 Evidence，最后写 Report 或是 Error），在海量并发的生产环境中，你怎么证明这三步是严丝合缝闭环的？
这里，我们要使用 Splunk 最为高级的 transaction 命令。它能通过共用的 session_id，将多条分散的日志“捏合”成一个完整的业务事务，并自动计算出整个链路的耗时！
请在 Splunk 中执行这段状态机校验语句：
index=main sourcetype="_json" (event_type="PEAK_Plan" OR event_type="PEAK_Evidence" OR event_type="PEAK_Final_Report" OR event_type="PEAK_Error")
| transaction session_id startswith=(event_type="PEAK_Plan") endswith=(event_type="PEAK_Final_Report" OR event_type="PEAK_Error")
| eval transaction_status = if(eventcount==3, "🟢 Healthy Closed Loop", if(eventcount==2 AND match(event_type, "PEAK_Error"), "🔴 Crashed Mid-flight", "🟡 Incomplete/Broken"))
| table _time, session_id, transaction_status, duration, eventcount, content.risk_score, error_message
| rename duration as "Total Execution Time (Sec)", eventcount as "Events in Transaction"
| sort - _time
🎯 你的验收指标：
1. transaction_status: 如果一切顺利，你会看到满屏的 🟢 Healthy Closed Loop。如果你故意制造了断网演练，你会看到 🔴 Crashed Mid-flight（意味着事务由 Plan 开始，但以 Error 终结）。
2. Events in Transaction: 健康状态下必须是 3（Plan + Evidence + Report）。
3. Total Execution Time (Sec): 这是 transaction 命令原生赋予的字段，它极其精准地测算出了从 AI 生成计划，到执行完毕最终定性，整个状态机流转花了多少秒。
🎉 Day 15 结语：看着前端整齐划一的 Transaction 闭环，身为架构师的你此刻可以放心地宣布：后端 Python 引擎正式封版！ 我们的数据基座已坚若磐石，它既能高并发运行，又具备完美的审计和容灾能力。
明天，也就是 Day 16，我们将进入整个项目的巅峰时刻：将这些冷冰冰的优质数据，转化为让 CISO（首席信息安全官）一眼惊艳的 企业级安全高管大屏 (SOC Executive Dashboard)！
👇 全套教程多平台同步更新 👇
https://blog.csdn.net/thewindrider/category_13144991.html
https://gitcode.com/Chang_feng_Po/Splunk-AI-PEAK-Tutorial