Catch new error patterns in the logs right after a deploy

id: post-deploy-log-pattern-diff
namespace: company.team

inputs:
  - id: service
    type: STRING
    defaults: checkout-api

  - id: scenario
    type: SELECT
    displayName: Demo scenario
    description: >
      CLEAN: the release logs the same patterns as before. NEW_ERROR: the
      release starts logging an error template nobody has seen before. SURGE: a
      known warning template becomes ten times more frequent.
    values:
      - CLEAN
      - NEW_ERROR
      - SURGE
    defaults: NEW_ERROR

  - id: surge_ratio
    type: FLOAT
    displayName: Surge ratio
    description: Flag a known template whose share of log lines grew by at least
      this factor after the deploy.
    defaults: 5.0

  - id: min_new_count
    type: INT
    displayName: Minimum count for a new pattern
    description: A new template must appear at least this many times after the
      deploy to count, so a single odd line does not block a release.
    defaults: 3

triggers:
  - id: after_deploy
    type: io.kestra.plugin.core.trigger.Webhook
    description: Call from the deploy pipeline about 15 minutes after a rollout.
      Replace the key first.
    key: replace-with-a-long-random-key

tasks:
  - id: diff_patterns
    type: io.kestra.plugin.scripts.python.Script
    description: >
      Mine log templates with Drain3 from the hour before the deploy, then feed
      the lines after the deploy into the same miner. Templates that appear only
      after the deploy are NEW. Known templates whose share of lines grew past
      surge_ratio are SURGING. Variable parts such as ids, durations and IPs are
      masked, so one message with different values is one template.
    containerImage: python:3.12-slim
    beforeCommands:
      - pip install -q --root-user-action=ignore drain3==0.9.11
    env:
      SCENARIO: "{{ inputs.scenario }}"
      SURGE: "{{ inputs.surge_ratio }}"
      MIN_NEW: "{{ inputs.min_new_count }}"
    outputFiles:
      - pattern_diff.md
    script: |
      import json, os, random
      from collections import Counter
      from drain3 import TemplateMiner
      from drain3.template_miner_config import TemplateMinerConfig
      from drain3.masking import MaskingInstruction

      random.seed(5)
      scenario, surge, min_new = os.environ["SCENARIO"], float(os.environ["SURGE"]), int(os.environ["MIN_NEW"])

      # Demo logs. Replace with an export from Loki, Elasticsearch or CloudWatch for the two windows.
      def ip(): return f"10.0.{random.randint(0, 255)}.{random.randint(1, 254)}"
      def normal_line():
          r = random.random()
          if r < 0.55: return f"GET /api/cart/{random.randint(1000, 99999)} 200 in {random.randint(3, 80)}ms from {ip()}"
          if r < 0.80: return f"POST /api/checkout order={random.randint(100000, 999999)} amount={random.randint(5, 900)}.{random.randint(0, 99):02d} status=ok"
          if r < 0.95: return f"cache hit ratio {random.randint(80, 99)}% for key cart:{random.randint(1000, 99999)}"
          return f"WARN payment provider slow response {random.randint(900, 3000)}ms for order={random.randint(100000, 999999)}"
      def after_line():
          if scenario == "NEW_ERROR" and random.random() < 0.02:
              return f"ERROR tax service returned null rate for region={random.choice(['EU-DE', 'EU-FR', 'US-CA'])} order={random.randint(100000, 999999)}"
          if scenario == "SURGE" and random.random() < 0.30:
              return f"WARN payment provider slow response {random.randint(900, 3000)}ms for order={random.randint(100000, 999999)}"
          return normal_line()
      before = [normal_line() for _ in range(6000)]
      after = [after_line() for _ in range(2000)]

      config = TemplateMinerConfig()
      config.drain_sim_th = 0.5
      config.masking_instructions = [
          MaskingInstruction(r"\d+\.\d+\.\d+\.\d+", "IP"),
          MaskingInstruction(r"(?<=[=/:])[A-Z]{2}-[A-Z]{2}\b", "REGION"),
          MaskingInstruction(r"\d+(\.\d+)?(ms|%)?", "NUM"),
      ]
      miner = TemplateMiner(config=config)
      seen_before = Counter(miner.add_log_message(l)["cluster_id"] for l in before)
      seen_after = Counter(miner.add_log_message(l)["cluster_id"] for l in after)
      templates = {c.cluster_id: c.get_template() for c in miner.drain.clusters}

      new, surging = [], []
      for cid, n in seen_after.most_common():
          share_after = n / len(after)
          if cid not in seen_before:
              if n >= min_new:
                  new.append((templates[cid], n))
          else:
              share_before = seen_before[cid] / len(before)
              if share_after / share_before >= surge:
                  surging.append((templates[cid], round(share_before * 100, 2), round(share_after * 100, 2)))

      verdict = "NEW_PATTERN" if new else ("SURGE" if surging else "CLEAN")
      lines = [f"# Log patterns after deploy: {verdict}", "",
               f"{len(before)} lines before, {len(after)} after, {len(templates)} templates.", "",
               "## New templates", *([f"- `{t}` ({n} lines)" for t, n in new] or ["- none"]), "",
               "## Surging templates", *([f"- `{t}`: {b}% to {a}% of lines" for t, b, a in surging] or ["- none"])]
      open("pattern_diff.md", "w").write("\n".join(lines) + "\n")
      print("\n".join(lines))
      print("::" + json.dumps({"outputs": {
          "verdict": verdict, "templates": len(templates), "new_count": len(new),
          "new": "; ".join(f"{t} ({n})" for t, n in new),
          "surging": "; ".join(f"{t} {b}% to {a}%" for t, b, a in surging)}}) + "::")

  - id: publish_diff
    type: io.kestra.plugin.core.namespace.UploadFiles
    namespace: "{{ flow.namespace }}"
    filesMap:
      "deploys/{{ inputs.service }}/log-patterns-{{ execution.id }}.md": "{{ outputs.diff_patterns.outputFiles['pattern_diff.md'] }}"

  - id: verdict
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.diff_patterns.vars.verdict }}"
    cases:
      CLEAN:
        - id: clean
          type: io.kestra.plugin.core.log.Log
          message: "{{ inputs.service }}: no new or surging log pattern after the deploy
            ({{ outputs.diff_patterns.vars.templates }} templates)."
      SURGE:
        - id: surge_warning
          type: io.kestra.plugin.core.log.Log
          level: WARN
          message: "{{ inputs.service }}: known patterns surged after the deploy: {{
            outputs.diff_patterns.vars.surging }}"
    defaults:
      - id: roll_back_signal
        type: io.kestra.plugin.core.execution.Fail
        description: A new error template right after a deploy is the strongest single
          signal that the release introduced it. Failing lets the deploy
          pipeline roll back.
        errorMessage: "{{ inputs.service }}: {{ outputs.diff_patterns.vars.new_count }}
          log pattern(s) appeared only after the deploy: {{
          outputs.diff_patterns.vars.new }}"