Skip to content

Instantly share code, notes, and snippets.

@Antonlovesdnb
Last active April 6, 2026 13:33
Show Gist options
  • Select an option

  • Save Antonlovesdnb/19c1ab49556eb6a4b46abff4270680bd to your computer and use it in GitHub Desktop.

Select an option

Save Antonlovesdnb/19c1ab49556eb6a4b46abff4270680bd to your computer and use it in GitHub Desktop.
detection-review
argument-hint <Obsidian note name>
allowed-tools
Bash
Read
Write

Detection Peer Review — Query Benchmarking & Optimization

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.

Input

$ARGUMENTS contains the Obsidian note name (without .md extension).

If no note name is provided, ask the user.

Configuration

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.

Workflow

Step 1 — Read note, extract query

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.

Step 2 — Benchmark original 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 seconds
  • resultCount — number of results
  • scanCount — number of events scanned
  • eventCount — 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.json

Step 3 — Analyze and optimize the query

Review the original query and generate an improved version. Apply these optimization checks in priority order:

3a. tstats conversion check

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.user
  • IpAddress / source IP → Authentication.src
  • EventCode → Authentication.signature_id
  • LogonType → Authentication.app
  • Failed logon filter → Authentication.action="failure"

CIM field mappings for Endpoint.Processes:

  • Image / process name → Processes.process_name
  • CommandLine → Processes.process
  • ParentImage → Processes.parent_process_name
  • User → Processes.user
  • Computer / 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).

3b. Early filtering

  • Move field filters as early as possible in the pipeline
  • If the query uses where after a stats, check if the filter can be moved before the stats as a search filter
  • Add index= and sourcetype= if missing (avoids scanning all indexes)
  • Add EventCode= filtering if the query targets a specific event type but doesn't filter on it

3c. Field extraction optimization

  • 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 count is cheaper

3d. Time range optimization

  • If the detection doesn't need 24h of data, suggest a tighter window
  • For high-volume indexes, suggest span bucketing in stats to reduce result cardinality

Step 4 — Benchmark optimized query

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.

Step 5 — Compare and validate

Compare:

  1. Performance: runDuration, scanCount, eventCount differences
  2. 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).

Step 6 — Write results back to Obsidian

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?>

Step 7 — Summary

Report to the user:

  • Original vs optimized run duration
  • Key optimization applied
  • Whether results matched
  • Confirm note updated
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment