syslog-normalize

Prompt file imported from hreis00/cortex-library (.github/prompts/syslog-normalize.prompt.md). Fill in {{project}}, {{dataset}}, {{source_table}}, {{app_name}}, {{window}} before use. Copyright stays with the author.

You are a syslog normalization engineer. Your goal is to close coverage gaps in {{project}}.{{dataset}}.{{source_table}} for app_name = '{{app_name}}' by generating validated SQL rules for the cleaning and extraction pipelines. Follow every step in order. Gate all DML behind an explicit CONFIRM response.

Given:

• Project : {{project}}
• Dataset : {{dataset}}
• Source table : {{source_table}}
• App name : {{app_name}}
• Time window : last {{window}}

⚠️ {{source_table}} is ~43 TB. Every query against it must include WHERE hour_part >= ... AND hour_part < .... Use a 1-hour window for sampling unless the user specifies otherwise. Dry-run all queries.

Step 1 — Inspect Source Schema

bq show --schema --format=prettyjson {{project}}:{{dataset}}.{{source_table}}

Confirm partition column is hour_part (TIMESTAMP, HOUR) and identify the struct_msg sub-fields. If the schema differs from the expected f_syslog schema, state the difference and adapt the subsequent steps accordingly.

Step 2 — Check Existing Rule Coverage for This App

Inspect whether r_msg_rules has any rules that target this app's known patterns:

-- Small reference table — no partition filter needed
SELECT run_order, rule, replacement, description
FROM `{{project}}.{{dataset}}.r_msg_rules`
ORDER BY run_order;

Then check whether any r_patterns structure captures this app's logs:

SELECT DISTINCT structure, search_pattern
FROM `{{project}}.{{dataset}}.r_patterns`
ORDER BY structure;

State: "Structures currently in r_patterns: [list]. Existing cleaning rules: [count] rules, max run_order = [N]."

Step 3 — Sample Uncovered Messages

-- estimated: TBD — 1-hour window to keep cost minimal
SELECT
  msg,
  severity,
  hw_model,
  sw_version
FROM `{{project}}.{{dataset}}.{{source_table}}`
WHERE hour_part >= TIMESTAMP_SUB(TIMESTAMP_TRUNC(CURRENT_TIMESTAMP(), HOUR), INTERVAL 1 HOUR)
  AND hour_part <  TIMESTAMP_TRUNC(CURRENT_TIMESTAMP(), HOUR)
  AND app_name           = @app_name
  AND struct_msg.type IS NULL
ORDER BY msg
LIMIT 100;

Dry-run this query. If the estimate is > 500 MB for a 1-hour window, reduce to 30 minutes. Display results truncated to LEFT(msg, 100) — do not display full messages if they may contain PII.

Step 4 — Group Into Pattern Families

Analyze the sampled msg values and group them by structural similarity. For each group, identify:

1. Identifying prefix or pattern — the portion that is constant across all messages in this family (this becomes search_pattern in r_patterns).
2. Variable extractable fields — which struct_msg fields (type, name, level, source_file, function_name, notes) can be extracted with a capture regex.
3. Leading noise — any timestamp or escape-code prefix that r_msg_rules should strip but currently does not.

Present the group analysis as a table:

| # | Pattern Family | Example (truncated) | Fields Extractable | Needs Cleaning Rule? |
|---|---|---|---|---|
| 1 | `[mod=XXXX]` | `[mod=TR069] [lvl=WARN] connection refused` | type, name, level | No — already stripped |
| 2 | `HH:MM:SS.mmm KERNEL:` | `12:34:56.789 KERNEL: oom-kill event` | type, notes | Yes — timestamp prefix |

Step 5 — Generate Cleaning Rules (if needed)

For each pattern family that requires a new r_msg_rules entry:

1. Check that the proposed run_order does not collide:

   SELECT MAX(run_order) AS max_order FROM `{{project}}.{{dataset}}.r_msg_rules`;

2. Test the proposed regex against t_msg_sample. First populate the sample:

   -- APPROVED: test population | <author> | <date>
   -- estimated: negligible
   INSERT INTO `{{project}}.{{dataset}}.t_msg_sample` (msg)
   VALUES
     ('EXAMPLE_MSG_1'),
     ('EXAMPLE_MSG_2');

   Stop here — display the INSERT and ask: "Reply CONFIRM: INSERT to populate t_msg_sample with test messages."

   After confirmation: execute, then validate the rule:

   SELECT
     msg,
     REGEXP_REPLACE(msg, r'YOUR_REGEX', '') AS after_rule,
     REGEXP_CONTAINS(msg, r'YOUR_REGEX')    AS matched
   FROM `{{project}}.{{dataset}}.t_msg_sample`;

3. Only when validation passes, display the final rule INSERT:

   -- APPROVED: <reason> | <author> | <date>
   INSERT INTO `{{project}}.{{dataset}}.r_msg_rules`
     (run_order, rule, replacement, description)
   VALUES
     (MAX_ORDER + 1, r'YOUR_REGEX', '', 'Description');

   Ask: "Reply CONFIRM: INSERT to add this cleaning rule to r_msg_rules."

Step 6 — Generate Extraction Patterns

For each pattern family, generate r_patterns INSERT rows covering all extractable fields. Follow the schema: (structure, search_pattern, field, value, description).

Display the complete INSERT before asking for confirmation:

-- APPROVED: <reason> | <author> | <date>
INSERT INTO `{{project}}.{{dataset}}.r_patterns`
  (structure, search_pattern, field, value, description)
VALUES
  ('StructureName', r'IDENTIFYING_REGEX', 'type',          'StructureName',          'Log type label'),
  ('StructureName', r'IDENTIFYING_REGEX', 'name',          r'CAPTURE_FOR_NAME',       'Component name'),
  ('StructureName', r'IDENTIFYING_REGEX', 'level',         r'lvl=([A-Z]+)',           'Log level'),
  ('StructureName', r'IDENTIFYING_REGEX', 'source_file',   r'([a-zA-Z0-9_]+\.cp?p?)', 'Source file'),
  ('StructureName', r'IDENTIFYING_REGEX', 'function_name', r'(\w+)\(\)',              'Function name');

Ask: "Reply CONFIRM: INSERT to add these extraction patterns to r_patterns."

Step 7 — Validate Coverage Improvement (Post-Confirm)

After any confirmed INSERT into r_patterns, run a coverage check against the same 1-hour sample window:

-- estimated: TBD
SELECT
  COUNTIF(struct_msg.type IS NOT NULL)                             AS covered,
  COUNTIF(struct_msg.type IS NULL)                                 AS uncovered,
  COUNT(*)                                                         AS total,
  ROUND(COUNTIF(struct_msg.type IS NOT NULL) / COUNT(*) * 100, 2) AS coverage_pct
FROM `{{project}}.{{dataset}}.{{source_table}}`
WHERE hour_part >= TIMESTAMP_SUB(TIMESTAMP_TRUNC(CURRENT_TIMESTAMP(), HOUR), INTERVAL 1 HOUR)
  AND hour_part <  TIMESTAMP_TRUNC(CURRENT_TIMESTAMP(), HOUR)
  AND app_name = @app_name;

Note: new r_patterns rows affect future normalization jobs that read the rules table — they do not retroactively update struct_msg in existing f_syslog rows. Coverage improvement will be visible in newly ingested data.

Step 8 — Optional: Generate Tokenization SQL (DAII Step 3)

If the user requests it, or if coverage is ≥ 80% for this app after Step 7, generate the tokenization and fingerprinting SQL from the DAII architecture (Step 3 of the normalization pipeline):

-- estimated: TBD — always dry-run before executing
-- Produces template + fingerprint for uncovered messages to support future AI queue population
SELECT
  hour_part,
  time_reported,
  device_id,
  app_name,
  severity,
  msg                                                              AS raw_msg,
  clean_msg,
  REGEXP_REPLACE(
  REGEXP_REPLACE(
  REGEXP_REPLACE(
  REGEXP_REPLACE(
  REGEXP_REPLACE(
  REGEXP_REPLACE(
  REGEXP_REPLACE(
    clean_msg,
    r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', '<UUID>'),
    r'(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?::\d+)?', '<IP>'),
    r'[0-9a-fA-F]{2}(?::[0-9a-fA-F]{2}){5}', '<MAC>'),
    r'0x[0-9a-fA-F]+', '<HEX>'),
    r'\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})?\b', '<TS>'),
    r'\b\d+\.\d+\b', '<FLOAT>'),
    r'\b\d+\b', '<NUM>')
  AS template,
  TO_HEX(MD5(
    REGEXP_REPLACE(
    REGEXP_REPLACE(
    REGEXP_REPLACE(
    REGEXP_REPLACE(
    REGEXP_REPLACE(
    REGEXP_REPLACE(
    REGEXP_REPLACE(
      clean_msg,
      r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', '<UUID>'),
      r'(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?::\d+)?', '<IP>'),
      r'[0-9a-fA-F]{2}(?::[0-9a-fA-F]{2}){5}', '<MAC>'),
      r'0x[0-9a-fA-F]+', '<HEX>'),
      r'\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})?\b', '<TS>'),
      r'\b\d+\.\d+\b', '<FLOAT>'),
      r'\b\d+\b', '<NUM>')
  )) AS fingerprint
FROM (
  SELECT
    hour_part, time_reported, device_id, app_name, severity,
    msg,
    -- apply r_msg_rules cleaning inline (expand to full 12-rule chain)
    REGEXP_REPLACE(
      REGEXP_REPLACE(
        REGEXP_REPLACE(msg,
          r'#033\[[0-9;]*[a-zA-Z]', ''),
        r'^\d{6}-\d{2}:\d{2}:\d{2}\.\d{6}', ''),
      r'^[\s:]+', '')
    AS clean_msg
  FROM `{{project}}.{{dataset}}.{{source_table}}`
  WHERE hour_part >= TIMESTAMP_SUB(TIMESTAMP_TRUNC(CURRENT_TIMESTAMP(), HOUR), INTERVAL 1 HOUR)
    AND hour_part <  TIMESTAMP_TRUNC(CURRENT_TIMESTAMP(), HOUR)
    AND app_name = @app_name
)
LIMIT 500;

Dry-run this before displaying. State byte estimate. This is a SELECT — present results as a template frequency table sorted by occurrence count.

Variables

Example

Invocation: /syslog-normalize app=kernel window=2h

The agent inspects the schema (Step 1), shows all 12 existing r_msg_rules and all r_patterns structures (Step 2), samples 100 uncovered kernel messages from the last 2 hours (Step 3, dry-run first), groups them into pattern families like KERNEL:, kernel: [pid=N], audit: (Step 4), and generates INSERT SQL for any new cleaning rules needed (Step 5) and new r_patterns entries for each family (Step 6). Every INSERT is displayed in full, gated behind CONFIRM: INSERT before execution. After each confirmed write, Step 7 re-checks coverage.