LaunchDarkly Flag Evaluation Request Spike Sentinel

id: launchdarkly-flag-evaluation-request-spike-sentinel
namespace: company.operations

description: |
  Audits LaunchDarkly feature flag configurations to detect archived flag evaluations,
  stale temporary flags, and unexpected evaluation surges before technical debt degrades client SDK performance.

concurrency:
  limit: 1

inputs:
  - id: project_key
    type: STRING
    displayName: LaunchDarkly Project Key
    defaults: "default"
    description: Key identifier of the target LaunchDarkly project.

  - id: environment_key
    type: STRING
    displayName: Environment Key
    defaults: "production"
    description: Key identifier of the target LaunchDarkly environment.

  - id: flag_key
    type: STRING
    displayName: Feature Flag Key
    defaults: "new-checkout-pipeline"
    description: Feature flag key to audit for evaluation status and lifecycle debt.

  - id: alert_on_anomaly
    type: BOOL
    displayName: Alert on Anomaly
    defaults: true
    description: Whether to dispatch automated Slack alerts when archived or stale
      feature flags are detected.

tasks:
  - id: evaluate_flag_status
    type: io.kestra.plugin.scripts.python.Script
    description: Queries LaunchDarkly REST API for feature flag configuration and
      lifecycle status.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      LAUNCHDARKLY_API_KEY: "{{ secret('LAUNCHDARKLY_API_KEY') }}"
    script: |
      import os
      import json
      import requests
      from kestra import Kestra

      project = "{{ inputs.project_key }}".strip()
      env = "{{ inputs.environment_key }}".strip()
      flag = "{{ inputs.flag_key }}".strip()
      token = os.environ.get("LAUNCHDARKLY_API_KEY", "")

      headers = {
          "Authorization": token,
          "LD-API-Version": "beta",
          "Accept": "application/json"
      }

      url = f"https://app.launchdarkly.com/api/v2/flags/{project}/{flag}"
      flag_name = flag
      is_archived = False
      is_temporary = False
      has_anomaly = False
      status_name = "unknown"

      try:
          resp = requests.get(url, headers=headers, timeout=15)
          resp.raise_for_status()
          flag_data = resp.json()

          flag_name = flag_data.get("name", flag)
          is_archived = flag_data.get("archived", False)
          is_temporary = flag_data.get("temporary", False)

          status_url = f"https://app.launchdarkly.com/api/v2/flag-statuses/{project}/{env}/{flag}"
          st_resp = requests.get(status_url, headers=headers, timeout=15)
          if st_resp.status_code == 200:
              st_data = st_resp.json()
              status_name = st_data.get("name", "active")

          # Anomaly condition: flag is archived but active or marked for cleanup
          if is_archived or (is_temporary and status_name in ["inactive", "completed"]):
              has_anomaly = True

          Kestra.outputs({
              "project": project,
              "environment": env,
              "flag_key": flag,
              "flag_name": flag_name,
              "is_archived": is_archived,
              "is_temporary": is_temporary,
              "status_name": status_name,
              "has_anomaly": has_anomaly
          })

      except Exception as e:
          print(f"Error querying LaunchDarkly API for flag {flag}: {e}")
          Kestra.outputs({
              "project": project,
              "environment": env,
              "flag_key": flag,
              "flag_name": flag_name,
              "is_archived": False,
              "is_temporary": False,
              "status_name": "error",
              "has_anomaly": False
          })

  - id: check_flag_anomaly
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether feature flag lifecycle triggers alerts for code
      deprecation.
    condition: "{{ outputs.evaluate_flag_status.vars.has_anomaly and
      inputs.alert_on_anomaly }}"
    then:
      - id: notify_slack_flag_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches feature flag lifecycle or evaluation anomaly warning to
          engineering on Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#feature-flags",
            "username": "LaunchDarkly-Sentinel",
            "text": "NOTICE: LaunchDarkly Flag `{{ outputs.evaluate_flag_status.vars.flag_name }}` (`{{ outputs.evaluate_flag_status.vars.flag_key }}`) in environment `{{ outputs.evaluate_flag_status.vars.environment }}` requires attention! Archived: {{ outputs.evaluate_flag_status.vars.is_archived }}, Temporary: {{ outputs.evaluate_flag_status.vars.is_temporary }}, Status: `{{ outputs.evaluate_flag_status.vars.status_name }}`. Console: https://app.launchdarkly.com/{{ outputs.evaluate_flag_status.vars.project }}/{{ outputs.evaluate_flag_status.vars.environment }}/features/{{ outputs.evaluate_flag_status.vars.flag_key }}"
          }

  - id: log_flag_status
    type: io.kestra.plugin.core.log.Log
    description: Records an operational audit trail of LaunchDarkly feature flag
      configurations.
    message: "LaunchDarkly audit completed for flag '{{
      outputs.evaluate_flag_status.vars.flag_key }}'. Status: {{
      outputs.evaluate_flag_status.vars.status_name }}, Archived: {{
      outputs.evaluate_flag_status.vars.is_archived }}, Anomaly: {{
      outputs.evaluate_flag_status.vars.has_anomaly }}."

  - id: export_flag_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured manifest of LaunchDarkly flag lifecycle audit for
      technical debt burndown.
    format:
      project: "{{ outputs.evaluate_flag_status.vars.project }}"
      environment: "{{ outputs.evaluate_flag_status.vars.environment }}"
      flag_key: "{{ outputs.evaluate_flag_status.vars.flag_key }}"
      flag_name: "{{ outputs.evaluate_flag_status.vars.flag_name }}"
      is_archived: "{{ outputs.evaluate_flag_status.vars.is_archived }}"
      is_temporary: "{{ outputs.evaluate_flag_status.vars.is_temporary }}"
      status: "{{ outputs.evaluate_flag_status.vars.status_name }}"
      has_anomaly: "{{ outputs.evaluate_flag_status.vars.has_anomaly }}"

triggers:
  - id: hourly_launchdarkly_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Automated schedule trigger auditing LaunchDarkly feature flag
      states hourly.
    cron: "0 * * * *"