Dynatrace Synthetic Browser Step Failure Sentinel

id: dynatrace-synthetic-browser-step-failure-sentinel
namespace: company.sre

description: |
  Audits Dynatrace Synthetic browser monitoring executions to detect multi-step user journey failures,
  DOM selector timeouts, and transaction errors before customer conversion funnels degrade.

concurrency:
  limit: 1

inputs:
  - id: dynatrace_tenant_uri
    type: STRING
    displayName: Dynatrace Tenant URI
    defaults: "https://abc12345.live.dynatrace.com"
    description: Base URL of your Dynatrace SaaS environment or Managed cluster endpoint.

  - id: synthetic_monitor_id
    type: STRING
    displayName: Synthetic Monitor ID
    defaults: "SYNTHETIC_TEST-1234567890ABCDEF"
    description: Dynatrace Synthetic Browser Monitor entity ID to audit for step failures.

  - id: max_failed_executions_threshold
    type: INT
    displayName: Max Failed Executions Threshold
    defaults: 1
    description: Maximum allowable failed executions in the recent evaluation window
      before triggering an incident.

  - id: check_window_minutes
    type: INT
    displayName: Check Window (Minutes)
    defaults: 15
    description: Rolling lookback duration in minutes for evaluating synthetic
      browser test executions.

  - id: alert_on_failure
    type: BOOL
    displayName: Alert on Failure
    defaults: true
    description: Whether to dispatch automated Slack incident alerts when synthetic
      browser step failures are detected.

tasks:
  - id: evaluate_dynatrace_browser_steps
    type: io.kestra.plugin.scripts.python.Script
    description: Queries Dynatrace Environment API v2 for synthetic browser monitor
      status and execution step telemetry.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      DYNATRACE_API_TOKEN: "{{ secret('DYNATRACE_API_TOKEN') }}"
    script: |
      import os
      import json
      import requests
      from datetime import datetime, timedelta, timezone
      from kestra import Kestra

      tenant = "{{ inputs.dynatrace_tenant_uri }}".rstrip('/')
      token = os.environ.get("DYNATRACE_API_TOKEN", "")
      monitor_id = "{{ inputs.synthetic_monitor_id }}".strip()
      threshold = int("{{ inputs.max_failed_executions_threshold }}")
      window_minutes = int("{{ inputs.check_window_minutes }}")

      headers = {
          "Authorization": f"Api-Token {token}",
          "Accept": "application/json"
      }

      now = datetime.now(timezone.utc)
      start_time = now - timedelta(minutes=window_minutes)
      start_ts_ms = int(start_time.timestamp() * 1000)

      monitor_name = "Unknown Monitor"
      monitor_enabled = True
      failed_executions_count = 0
      recent_failures = []
      has_failures = False

      try:
          mon_resp = requests.get(f"{tenant}/api/v2/synthetic/monitors/{monitor_id}", headers=headers, timeout=15)
          if mon_resp.status_code == 200:
              mon_data = mon_resp.json()
              monitor_name = mon_data.get("name", monitor_id)
              monitor_enabled = mon_data.get("enabled", True)

          exec_resp = requests.get(
              f"{tenant}/api/v2/synthetic/execution/{monitor_id}",
              headers=headers,
              params={"from": start_ts_ms},
              timeout=15
          )

          if exec_resp.status_code == 200:
              exec_data = exec_resp.json()
              executions = exec_data.get("executions", [])

              for execution in executions:
                  result = execution.get("result", "UNKNOWN")
                  if result == "FAILED":
                      failed_executions_count += 1
                      recent_failures.append({
                          "execution_id": execution.get("executionId"),
                          "location": execution.get("location"),
                          "timestamp": execution.get("timestamp"),
                          "step_details": execution.get("stepResults", [])
                      })

          if failed_executions_count >= threshold and monitor_enabled:
              has_failures = True

          Kestra.outputs({
              "monitor_name": monitor_name,
              "monitor_id": monitor_id,
              "monitor_enabled": monitor_enabled,
              "failed_executions_count": failed_executions_count,
              "has_failures": has_failures,
              "failures_details": json.dumps(recent_failures[:5])
          })

      except Exception as e:
          print(f"Error querying Dynatrace Synthetic API: {e}")
          Kestra.outputs({
              "monitor_name": monitor_name,
              "monitor_id": monitor_id,
              "monitor_enabled": False,
              "failed_executions_count": 0,
              "has_failures": False,
              "failures_details": "[]"
          })

  - id: check_synthetic_failure
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether Dynatrace synthetic test executions recorded step
      failures exceeding threshold.
    condition: "{{ outputs.evaluate_dynatrace_browser_steps.vars.has_failures and
      inputs.alert_on_failure }}"
    then:
      - id: notify_slack_synthetic_failure
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority incident notifications to Slack with
          failed synthetic journey details.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#sre-synthetic-alerts",
            "username": "Dynatrace-Sentinel",
            "text": "CRITICAL: Dynatrace Synthetic Browser Monitor `{{ outputs.evaluate_dynatrace_browser_steps.vars.monitor_name }}` (`{{ outputs.evaluate_dynatrace_browser_steps.vars.monitor_id }}`) recorded {{ outputs.evaluate_dynatrace_browser_steps.vars.failed_executions_count }} failed execution(s) in the last {{ inputs.check_window_minutes }} minutes. Deep link: {{ inputs.dynatrace_tenant_uri }}/#syntheticmonitor;id={{ outputs.evaluate_dynatrace_browser_steps.vars.monitor_id }}"
          }

  - id: log_synthetic_status
    type: io.kestra.plugin.core.log.Log
    description: Records an operational audit trail of Dynatrace synthetic browser
      monitoring evaluation results.
    message: "Dynatrace Synthetic audit completed for '{{
      outputs.evaluate_dynatrace_browser_steps.vars.monitor_name }}'. Failed
      executions: {{
      outputs.evaluate_dynatrace_browser_steps.vars.failed_executions_count }},
      Has Failures: {{
      outputs.evaluate_dynatrace_browser_steps.vars.has_failures }}."

  - id: export_dynatrace_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured manifest of synthetic execution results for
      observability and reliability dashboards.
    format:
      tenant: "{{ inputs.dynatrace_tenant_uri }}"
      monitor_id: "{{ outputs.evaluate_dynatrace_browser_steps.vars.monitor_id }}"
      monitor_name: "{{ outputs.evaluate_dynatrace_browser_steps.vars.monitor_name }}"
      failed_executions: "{{
        outputs.evaluate_dynatrace_browser_steps.vars.failed_executions_count
        }}"
      has_failures: "{{ outputs.evaluate_dynatrace_browser_steps.vars.has_failures }}"
      details: "{{ outputs.evaluate_dynatrace_browser_steps.vars.failures_details }}"

triggers:
  - id: fifteen_minute_dynatrace_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Automated schedule trigger auditing Dynatrace Synthetic browser
      runs every fifteen minutes.
    cron: "*/15 * * * *"