| argument-hint | <Obsidian note name> | |||
|---|---|---|---|---|
| allowed-tools |
|
Read a Splunk detection query from an Obsidian note, benchmark it, analyze for improvements (tstats conversion, field extraction, early filtering), run the optimized version, and compare performance. Results are written back to the note.
$ARGUMENTS contains the Obsidian note name (without .md extension).
If no note name is provided, ask the user.
SPLUNK_URL — from $SPLUNK_URL
SPLUNK_TOKEN — from $SPLUNK_TOKEN
VAULT=Obsidian Vault
Verify both env vars exist. If token auth fails (HTTP 401 or "not properly authenticated"), fall back to basic auth using $SPLUNK_USER and $SPLUNK_PASS env vars, or -u user:pass as a last resort.
obsidian read vault="{VAULT}" file="{NOTE_NAME}"Find the Splunk query in the note. Look for a splunk or spl fenced code block. If none found, look for any code block containing index= or | tstats or other SPL patterns. Extract the raw query.
Run the query as a normal job (not oneshot) to get job inspector stats:
# Create the job
JOB_SID=$(curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "search=search <ORIGINAL_QUERY>" \
-d "earliest_time=-24h" \
-d "latest_time=now" \
-d "exec_mode=normal" \
-d "output_mode=json" \
"$SPLUNK_URL/services/search/jobs" | python3 -c "import sys,json; print(json.load(sys.stdin).get('sid',''))")
# Poll until done (check every 2 seconds, max 120s)To poll, check the job status:
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
"$SPLUNK_URL/services/search/jobs/$JOB_SID?output_mode=json"Look for "dispatchState": "DONE" in the response content.
Once done, get the performance stats:
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
"$SPLUNK_URL/services/search/jobs/$JOB_SID?output_mode=json"Extract from the response content:
runDuration— total wall-clock time in secondsresultCount— number of resultsscanCount— number of events scannedeventCount— number of events matched
Also get the results:
curl -s -k \
-H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "output_mode=json" \
-d "count=50" \
"$SPLUNK_URL/services/search/jobs/$JOB_SID/results" > /tmp/detection_original_results.jsonReview the original query and generate an improved version. Apply these optimization checks in priority order:
tstats queries run against pre-built summaries (tsidx or accelerated data models) and are typically 5-100x faster than raw search. Always attempt a tstats conversion.
Step 3a.1 — Check for accelerated data models
First, check which CIM data models are accelerated:
curl -s -k -H "Authorization: Bearer $SPLUNK_TOKEN" \
-d "search=| rest /servicesNS/-/-/data/models | where acceleration=1 | table title" \
-d "output_mode=json" -d "exec_mode=oneshot" \
"$SPLUNK_URL/services/search/jobs/export"Map the original query's intent to a CIM data model:
- EventCode 4625 (failed logon) → Authentication data model,
Authentication.action="failure" - EventCode 4624 (successful logon) → Authentication data model,
Authentication.action="success" - EventCode 1 (process creation) → Endpoint.Processes data model
- EventCode 3 (network connection) → Network_Traffic data model
- EventCode 4688 (process creation) → Endpoint.Processes data model
- EventCode 7045 (service install) → Change data model
Step 3a.2 — Build the tstats query
If a matching accelerated data model exists, build a tstats query using CIM field names:
| tstats count from datamodel=<ModelName>.<ObjectName>
where <filter conditions using CIM fields>
by <CIM fields>
| rename <CIM field> as <friendly name> ...
CIM field mappings for Authentication:
TargetUserName→Authentication.userIpAddress/ source IP →Authentication.srcEventCode→Authentication.signature_idLogonType→Authentication.app- Failed logon filter →
Authentication.action="failure"
CIM field mappings for Endpoint.Processes:
Image/ process name →Processes.process_nameCommandLine→Processes.processParentImage→Processes.parent_process_nameUser→Processes.userComputer/ host →Processes.dest
Step 3a.3 — Note tradeoffs
Some original fields may not have CIM equivalents (e.g., NTLM Status/Sub_Status codes are not in the Authentication data model). Document what's gained (speed) and what's lost (field granularity). If critical fields are missing from the tstats version, also provide a hybrid approach:
| tstats count from datamodel=Authentication.Authentication
where Authentication.action="failure" Authentication.signature_id=4625
by Authentication.user Authentication.src
| rename Authentication.user as TargetUserName Authentication.src as IpAddress
Step 3a.4 — Fallback: raw tstats against indexed fields
If no data model matches, attempt tstats against raw indexed fields:
| tstats count where index=<index> EventCode=<code> by host sourcetype source
This only works with indexed fields (host, source, sourcetype, index, _time, and any fields in INDEXED_EXTRACTIONS or TRANSFORMS with WRITE_META=true).
- Move field filters as early as possible in the pipeline
- If the query uses
whereafter astats, check if the filter can be moved before the stats as asearchfilter - Add
index=andsourcetype=if missing (avoids scanning all indexes) - Add
EventCode=filtering if the query targets a specific event type but doesn't filter on it
- If the query only needs specific fields, add
| fields <field1> <field2>early to reduce memory usage - Replace
*wildcards in field values with specific values where possible - For
stats count(field)— if the field is always present,stats countis cheaper
- If the detection doesn't need 24h of data, suggest a tighter window
- For high-volume indexes, suggest
spanbucketing instatsto reduce result cardinality
Run the optimized query using the same job-based approach as Step 2. Use the exact same time range (earliest_time and latest_time).
Save results to /tmp/detection_optimized_results.json.
Compare:
- Performance: runDuration, scanCount, eventCount differences
- Result correctness: Verify the optimized query returns the same (or equivalent) results. Compare result counts and spot-check values.
If results differ, note the discrepancy and explain why (e.g., tstats counts may differ slightly from raw search due to indexing timing).
Append a ## Detection Peer Review section to the original note:
obsidian read vault="{VAULT}" file="{NOTE_NAME}"Then use obsidian update or obsidian create with overwrite to write the updated note.
The appended section should contain:
## Detection Peer Review — YYYY-MM-DD
### Original Query
\`\`\`splunk
<original query>
\`\`\`
### Performance Benchmark (Original)
| Metric | Value |
|--------|-------|
| Run Duration | Xs |
| Events Scanned | N |
| Events Matched | N |
| Results | N |
### Analysis
<Brief explanation of what the query does, what it detects, and what optimizations are possible. Be specific about WHY each optimization helps.>
### Optimized Query
\`\`\`splunk
<optimized query>
\`\`\`
### Performance Benchmark (Optimized)
| Metric | Value |
|--------|-------|
| Run Duration | Xs |
| Events Scanned | N |
| Events Matched | N |
| Results | N |
### Comparison
\`\`\`
Metric Original Optimized Improvement
─────────────────────────────────────────────────────
Duration X.Xs X.Xs X.X% faster
Events Scanned N N X.X% fewer
Results N N match/differ
\`\`\`
### Verdict
<One-paragraph summary: is the optimized query better? Are results equivalent? Should the detection be updated?>Report to the user:
- Original vs optimized run duration
- Key optimization applied
- Whether results matched
- Confirm note updated