Telemetry Invariant Scan #676
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| name: Telemetry Invariant Scan | |
| # Hourly invariant scan over prod-checkpoint-telemetry-events. | |
| # | |
| # WHY: this workflow exists to detect telemetry-classification leaks | |
| # *automatically*, instead of relying on humans to eyeball digest emails | |
| # 10+ days late (the May 2026 pattern that produced three cleanup-status | |
| # cohorts totalling 1248+ polluted rows). Every leak class becomes a | |
| # named invariant in ee/platform/telemetry-filter/invariants.json; this | |
| # workflow runs each one hourly and pages on any nonzero count. | |
| # | |
| # WHAT: scripts/telemetry-invariant-scan.py queries each invariant via | |
| # DDB GSI Query (read-only), publishes a CloudWatch custom metric per | |
| # invariant under namespace AxonFlow/TelemetryHealth, and exits non-zero | |
| # if any invariant's count exceeds its declared tolerance. On exit-2: | |
| # (a) a GitHub issue is opened with the invariant names + sample rows, | |
| # (b) an SNS alert fires on the existing community-saas-telemetry-alerts | |
| # topic — same alert pipeline as the DLQ + ingest-error alarms (memory | |
| # feedback_redeploy_all_writer_services_sharing_sink.md). | |
| # | |
| # WARN vs STRICT MODE: | |
| # - workflow_dispatch with enforce=warn → never fails the workflow. | |
| # Used during initial deployment + during cleanup-replay windows so | |
| # ongoing PR work isn't blocked by known historic violations. | |
| # - cron schedule → enforce=strict (default). After the 2026-05-19 | |
| # one-shot replay drops all counts to 0, strict mode is the steady | |
| # state. New violations fire the alarm next hour. | |
| # | |
| # DEFENSE-IN-DEPTH: CloudWatch metric publish happens regardless of | |
| # enforce mode, so a CW alarm can fire even if the workflow itself is | |
| # misconfigured. The GitHub issue + SNS alert are the workflow-side | |
| # response; the CW alarm is the AWS-side response. | |
| on: | |
| schedule: | |
| # Hourly, at :17 past the hour. Off-cycle from the digest cron (XX:15) | |
| # so the two don't share queue time on the runner pool. | |
| - cron: '17 * * * *' | |
| workflow_dispatch: | |
| inputs: | |
| enforce: | |
| description: 'strict (cron default) = exit 2 on any invariant failure. warn = always exit 0 with metric still published. Use warn during cleanup replay so in-progress work doesn''t fail unrelated PRs.' | |
| required: false | |
| type: choice | |
| default: strict | |
| options: | |
| - strict | |
| - warn | |
| table: | |
| description: 'DDB table to scan (default prod-checkpoint-telemetry-events). Override only for testing — staging/dev tables have their own scanners if needed.' | |
| required: false | |
| type: string | |
| default: prod-checkpoint-telemetry-events | |
| concurrency: | |
| # Only one scan at a time across the org. Scan is ~30s; back-pressure | |
| # from a long-running scan blocking the next cron is acceptable | |
| # behaviour (skip the next cron) — alternative would be parallel | |
| # publishes producing inconsistent CW metric timestamps. | |
| group: telemetry-invariant-scan | |
| cancel-in-progress: false | |
| permissions: | |
| contents: read | |
| issues: write # Open a GH issue on failure (severity-tagged) | |
| id-token: write # OIDC for AWS | |
| env: | |
| AWS_REGION: us-east-1 | |
| jobs: | |
| scan: | |
| name: Scan telemetry invariants | |
| runs-on: ubuntu-latest | |
| timeout-minutes: 10 | |
| outputs: | |
| failed_count: ${{ steps.scan.outputs.failed_count }} | |
| summary_path: ${{ steps.scan.outputs.summary_path }} | |
| steps: | |
| - name: Checkout | |
| uses: actions/checkout@v6 | |
| - name: Setup Python | |
| uses: actions/setup-python@v6 | |
| with: | |
| python-version: '3.11' | |
| - name: Install boto3 | |
| run: | | |
| python -m pip install --upgrade pip | |
| pip install boto3 | |
| - name: Configure AWS credentials | |
| uses: aws-actions/configure-aws-credentials@v6 | |
| with: | |
| aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID_INTERNAL }} | |
| aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY_INTERNAL }} | |
| aws-region: ${{ env.AWS_REGION }} | |
| - name: Run invariant scan | |
| id: scan | |
| env: | |
| ENFORCE: ${{ inputs.enforce || 'strict' }} | |
| TABLE: ${{ inputs.table || 'prod-checkpoint-telemetry-events' }} | |
| run: | | |
| set +e # we explicitly handle scanner's exit code; don't propagate | |
| SUMMARY_PATH="${RUNNER_TEMP}/invariant-scan-summary.json" | |
| python3 scripts/telemetry-invariant-scan.py \ | |
| --table "$TABLE" \ | |
| --region "$AWS_REGION" \ | |
| --enforce "$ENFORCE" \ | |
| --summary-json > "$SUMMARY_PATH" | |
| SCAN_EXIT=$? | |
| set -e | |
| # Count failures from the JSON summary regardless of enforce | |
| # mode — the metric publish + issue-on-failure pipeline runs | |
| # on FAILURE_COUNT > 0, not on exit code. This way warn-mode | |
| # still produces actionable signal. | |
| FAILED_COUNT=$(python3 -c "import json,sys; d=json.load(open(sys.argv[1])); print(d.get('failed_count', 0))" "$SUMMARY_PATH") | |
| echo "failed_count=$FAILED_COUNT" >> "$GITHUB_OUTPUT" | |
| echo "summary_path=$SUMMARY_PATH" >> "$GITHUB_OUTPUT" | |
| echo "scan_exit=$SCAN_EXIT" >> "$GITHUB_OUTPUT" | |
| echo "Scan summary:" | |
| cat "$SUMMARY_PATH" | |
| # Re-raise scanner's exit code so the workflow surfaces strict-mode | |
| # failures (the issue-creation + SNS-alert steps are gated on | |
| # FAILED_COUNT > 0 and run regardless of this exit). | |
| exit $SCAN_EXIT | |
| - name: Open GitHub issue on invariant violation | |
| # Run on FAILED_COUNT > 0 regardless of enforce mode. The issue is | |
| # the same regardless of whether enforce=strict (workflow fails) | |
| # or enforce=warn (workflow passes) — issue carries the diagnostic. | |
| # always() needed because strict-mode scanner exits non-zero and | |
| # default GHA semantics skip subsequent steps; we want this to run. | |
| if: ${{ always() && steps.scan.outputs.failed_count != '0' && steps.scan.outputs.failed_count != '' }} | |
| env: | |
| GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} | |
| FAILED_COUNT: ${{ steps.scan.outputs.failed_count }} | |
| SUMMARY_PATH: ${{ steps.scan.outputs.summary_path }} | |
| run: | | |
| set -euo pipefail | |
| # Construct a markdown body listing each failing invariant with | |
| # its sample rows + linked classifier rule. The body is the | |
| # triage starting point — operator should be able to land on | |
| # the relevant filters.json rule + replay command in one read. | |
| BODY_PATH="${RUNNER_TEMP}/issue-body.md" | |
| python3 - "$SUMMARY_PATH" "$BODY_PATH" <<'PY' | |
| import json, sys | |
| summary = json.load(open(sys.argv[1])) | |
| body_path = sys.argv[2] | |
| lines = [ | |
| f"# Telemetry invariant scan — {summary.get('failed_count', 0)} failing", | |
| "", | |
| f"Scanned table: `{summary.get('table')}`", | |
| f"Invariants config version: `{summary.get('invariants_version')}`", | |
| "", | |
| "## Failing invariants", | |
| "", | |
| ] | |
| for r in summary.get("results", []): | |
| if r.get("status") == "PASS": | |
| continue | |
| lines += [ | |
| f"### `{r['name']}` ({r['severity']})", | |
| "", | |
| f"- **Count:** {r['count']} (tolerance {r['tolerance']})", | |
| f"- **Description:** {r['description']}", | |
| f"- **Linked writer rule:** `{r['linked_writer_rule']}`", | |
| "", | |
| "**Sample violating rows:**", | |
| "", | |
| "```json", | |
| json.dumps(r.get("samples", []), indent=2), | |
| "```", | |
| "", | |
| "**Next steps (per the playbook):**", | |
| "", | |
| "1. Confirm the linked writer rule is correct — if not, fix `ee/platform/telemetry-filter/filters.json` + classify.go.", | |
| f"2. Replay the affected rows: `python3 scripts/replay-classifier-historic.py --invariant {r['name']} --apply`", | |
| "3. Re-run this scan; this invariant's count must return to 0.", | |
| "", | |
| ] | |
| open(body_path, "w").write("\n".join(lines)) | |
| PY | |
| ISSUE_TITLE="telemetry: $FAILED_COUNT invariant(s) failing on prod-checkpoint-telemetry-events" | |
| # Reuse existing issue if one is already open for this title (dedup | |
| # against hourly cron creating identical issues until the | |
| # violations are cleared). | |
| EXISTING=$(gh issue list \ | |
| --search "$ISSUE_TITLE in:title is:open" \ | |
| --json number --jq '.[0].number // empty' || true) | |
| if [ -n "$EXISTING" ]; then | |
| echo "::notice::Existing issue #$EXISTING — appending diagnostic comment instead of creating duplicate." | |
| gh issue comment "$EXISTING" --body-file "$BODY_PATH" | |
| else | |
| gh issue create \ | |
| --title "$ISSUE_TITLE" \ | |
| --label telemetry-health \ | |
| --body-file "$BODY_PATH" || \ | |
| gh issue create \ | |
| --title "$ISSUE_TITLE" \ | |
| --body-file "$BODY_PATH" # fallback if label doesn't exist | |
| fi | |
| - name: Publish SNS alert on invariant violation | |
| if: ${{ always() && steps.scan.outputs.failed_count != '0' && steps.scan.outputs.failed_count != '' }} | |
| env: | |
| FAILED_COUNT: ${{ steps.scan.outputs.failed_count }} | |
| SUMMARY_PATH: ${{ steps.scan.outputs.summary_path }} | |
| run: | | |
| # Reuse the existing community-saas-telemetry-alerts SNS topic | |
| # (set up in epic #2047 P5b for SoX-stack DLQ + ingest-error alarms). | |
| # Falls back to a no-op if the topic isn't resolvable rather than | |
| # failing the workflow — the GH issue already carries the | |
| # diagnostic; SNS is the paging-channel signal. | |
| TOPIC_ARN=$(aws sns list-topics --region "$AWS_REGION" \ | |
| --query "Topics[?contains(TopicArn, ':community-saas-telemetry-alerts')].TopicArn | [0]" \ | |
| --output text || echo "None") | |
| if [ "$TOPIC_ARN" = "None" ] || [ -z "$TOPIC_ARN" ]; then | |
| echo "::warning::community-saas-telemetry-alerts SNS topic not found — skipping alert publish. The GH issue still carries the diagnostic." | |
| exit 0 | |
| fi | |
| SUBJECT="telemetry-invariant-scan: $FAILED_COUNT failing" | |
| MESSAGE=$(python3 -c " | |
| import json,sys | |
| d=json.load(open(sys.argv[1])) | |
| out=[f'{d.get(\"failed_count\",0)} invariant(s) failing on {d.get(\"table\")} at $(date -u +%FT%TZ).', ''] | |
| for r in d.get('results',[]): | |
| if r.get('status')!='PASS': | |
| out.append(f' - {r[\"name\"]}: count={r[\"count\"]} (severity {r[\"severity\"]})') | |
| out.append('') | |
| out.append(f'GH Actions run: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}') | |
| print('\n'.join(out)) | |
| " "$SUMMARY_PATH") | |
| aws sns publish --region "$AWS_REGION" \ | |
| --topic-arn "$TOPIC_ARN" \ | |
| --subject "$SUBJECT" \ | |
| --message "$MESSAGE" | |
| - name: Upload scan summary as artifact | |
| if: ${{ always() }} | |
| uses: actions/upload-artifact@v7 | |
| with: | |
| name: invariant-scan-summary-${{ github.run_id }} | |
| path: ${{ steps.scan.outputs.summary_path }} | |
| retention-days: 30 |