flow-deploy.yml

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
name: Flow Deploy

on:
  issue_comment:
    types: [created]

permissions:
  contents: read
  pull-requests: write
  checks: write

jobs:
  detect:
    if: >
      github.event.issue.pull_request &&
      (contains(github.event.comment.body, 'deploy this flow') ||
       contains(github.event.comment.body, 'deploy the flow'))
    runs-on: ubuntu-latest
    outputs:
      items: ${{ steps.detect.outputs.items }}
      has_flows: ${{ steps.detect.outputs.has_flows }}
      ref: ${{ steps.pr-info.outputs.ref }}
      sha: ${{ steps.pr-info.outputs.sha }}
    steps:
      - name: Check Repository Permission
        uses: actions/github-script@v9
        with:
          script: |
            const user = context.payload.comment.user.login;
            const { data: permission } = await github.rest.repos.getCollaboratorPermissionLevel({
              owner: context.repo.owner,
              repo: context.repo.repo,
              username: user
            });
            const allowed = ['admin', 'maintain'].includes(permission.permission);
            if (!allowed) {
              core.setFailed(`User ${user} does not have admin/maintain permission.`);
            }

      - name: Get PR Info
        uses: actions/github-script@v9
        id: pr-info
        with:
          script: |
            const { data: pr } = await github.rest.pulls.get({
              owner: context.repo.owner,
              repo: context.repo.repo,
              pull_number: context.issue.number
            });
            core.setOutput('ref', pr.head.ref);
            core.setOutput('sha', pr.head.sha);

      - name: Checkout PR Branch
        uses: actions/checkout@v6
        with:
          ref: ${{ steps.pr-info.outputs.ref }}

      - name: Detect Flows and Test Configs
        uses: actions/github-script@v9
        id: detect
        with:
          script: |
            const fs = require('fs');
            const path = require('path');
            const { data: files } = await github.rest.pulls.listFiles({
              owner: context.repo.owner,
              repo: context.repo.repo,
              pull_number: context.issue.number,
              per_page: 100
            });
            const flows = files
              .filter(f => f.filename.startsWith('flows/') && f.filename.endsWith('.json'))
              .filter(f => f.status !== 'removed')
              .map(f => f.filename);

            if (flows.length === 0) {
              core.setOutput('has_flows', 'false');
              core.setOutput('items', '[]');
              return;
            }

            const items = [];
            for (const flowPath of flows) {
              const flowDir = path.dirname(flowPath);
              const flowBase = path.basename(flowPath, '.json');
              const testName = flowBase.replace(/-/g, '_');
              const testYaml = path.join(flowDir, 'tests', `test_${testName}.yaml`);

              let githubEnv = '';
              if (fs.existsSync(testYaml)) {
                const yaml = fs.readFileSync(testYaml, 'utf8');
                const envMatch = yaml.match(/github_environment:\s*(\S+)/);
                if (envMatch) githubEnv = envMatch[1];
              }

              if (!githubEnv) {
                core.warning(`No test YAML or github_environment for ${flowPath} — skipping`);
                continue;
              }

              items.push({
                flow_path: flowPath,
                test_yaml: testYaml,
                github_env: githubEnv,
                flow_base: flowBase,
              });
            }

            core.setOutput('has_flows', items.length > 0 ? 'true' : 'false');
            core.setOutput('items', JSON.stringify(items));
            core.info(`Found ${items.length} flow(s) with test configs`);

  deploy:
    needs: detect
    if: needs.detect.outputs.has_flows == 'true'
    runs-on: ubuntu-latest
    strategy:
      max-parallel: 1
      matrix:
        item: ${{ fromJSON(needs.detect.outputs.items) }}
    environment: ${{ matrix.item.github_env }}
    name: "Deploy: ${{ matrix.item.flow_base }}"

    env:
      SNOWFLAKE_ACCOUNT_URL: ${{ vars.SNOWFLAKE_ACCOUNT_URL }}
      SNOWFLAKE_USER: ${{ vars.SNOWFLAKE_USER }}
      SNOWFLAKE_PAT: ${{ secrets.SNOWFLAKE_PAT }}
      SNOWFLAKE_ROLE: ${{ vars.SNOWFLAKE_ROLE }}
      NIFI_RUNTIME_PAT: ${{ secrets.NIFI_RUNTIME_PAT }}
      NIFIHUB_REGISTRY_PAT: ${{ secrets.NIFIHUB_REGISTRY_PAT }}
      GH_SECRETS_JSON: ${{ toJSON(secrets) }}
      GH_VARS_JSON: ${{ toJSON(vars) }}
      GITHUB_TOKEN: ${{ github.token }}

    steps:
      - name: Create Check Run
        uses: actions/github-script@v9
        id: check
        with:
          script: |
            const { data: check } = await github.rest.checks.create({
              owner: context.repo.owner,
              repo: context.repo.repo,
              name: 'Flow Deploy Tests',
              head_sha: '${{ needs.detect.outputs.sha }}',
              status: 'in_progress',
              started_at: new Date().toISOString(),
              output: {
                title: 'Flow Deploy Tests',
                summary: `Deploying \`${{ matrix.item.flow_path }}\` to ephemeral runtime...`
              }
            });
            core.setOutput('id', check.id);

      - name: Checkout PR Branch
        uses: actions/checkout@v6
        with:
          ref: ${{ needs.detect.outputs.ref || github.event.issue.pull_request.head.ref }}
          fetch-depth: 0

      - name: Set up Python
        uses: actions/setup-python@v6
        with:
          python-version: '3.12'

      - name: Install Snowflake CLI
        uses: snowflakedb/snowflake-cli-action@v2.0

      - name: Install dependencies
        run: |
          pip install -r scripts/cd/requirements-cd.txt
          pip install -r flows/requirements.txt

      - name: Generate Runtime Name
        id: name
        run: |
          FLOW_BASE="${{ matrix.item.flow_base }}"
          PR_NUM="${{ github.event.issue.number }}"
          RUN_ID="${{ github.run_id }}"
          RUNTIME_NAME="CI_$(echo "$FLOW_BASE" | tr '[:lower:]-' '[:upper:]_')_${PR_NUM}_${RUN_ID}"
          RUNTIME_NAME="${RUNTIME_NAME:0:254}"
          echo "runtime_name=$RUNTIME_NAME" >> "$GITHUB_OUTPUT"
          echo "Runtime name: $RUNTIME_NAME"

      - name: Provision Ephemeral Runtime
        id: provision
        run: |
          URL_FILE=$(mktemp)
          ERROR_LOG=$(mktemp)
          echo "error_log=$ERROR_LOG" >> "$GITHUB_OUTPUT"
          python scripts/ci/provision_ci_runtime.py \
            --config "${{ matrix.item.test_yaml }}" \
            --runtime-name "${{ steps.name.outputs.runtime_name }}" \
            --output-file "$URL_FILE" 2>&1 | tee "$ERROR_LOG"
          PROVISION_EXIT=${PIPESTATUS[0]}
          if [ $PROVISION_EXIT -ne 0 ]; then
            exit $PROVISION_EXIT
          fi
          RUNTIME_URL=$(cat "$URL_FILE")
          echo "runtime_url=$RUNTIME_URL" >> "$GITHUB_OUTPUT"
          echo "Runtime URL: $RUNTIME_URL"

      - name: Deploy Flow
        id: deploy
        run: |
          PG_ID_FILE=$(mktemp)
          ERROR_LOG=$(mktemp)
          echo "error_log=$ERROR_LOG" >> "$GITHUB_OUTPUT"
          python scripts/ci/deploy_ci_flow.py \
            --flow-path "${{ matrix.item.flow_path }}" \
            --config "${{ matrix.item.test_yaml }}" \
            --runtime-url "${{ steps.provision.outputs.runtime_url }}" \
            --pat "$NIFI_RUNTIME_PAT" \
            --output-file "$PG_ID_FILE" 2>&1 | tee "$ERROR_LOG"
          DEPLOY_EXIT=${PIPESTATUS[0]}
          if [ $DEPLOY_EXIT -ne 0 ]; then
            exit $DEPLOY_EXIT
          fi
          PG_ID=$(cat "$PG_ID_FILE")
          echo "pg_id=$PG_ID" >> "$GITHUB_OUTPUT"
          echo "Deployed PG: $PG_ID"

      - name: Wait for Flow Execution
        if: always() && steps.deploy.outcome == 'success'
        run: |
          echo "Waiting 60 seconds for flow to process data..."
          sleep 60

      - name: Run Flow Tests
        if: always() && steps.deploy.outcome == 'success'
        id: tests
        env:
          DEPLOYED_PG_ID: ${{ steps.deploy.outputs.pg_id }}
          SNOWFLAKE_RUNTIME_URL: ${{ steps.provision.outputs.runtime_url }}
          SNOWFLAKE_RUNTIME_PAT: ${{ secrets.NIFI_RUNTIME_PAT }}
        run: |
          FLOW_PATH="${{ matrix.item.flow_path }}"
          FLOW_DIR=$(dirname "$FLOW_PATH")
          FLOW_BASE=$(basename "$FLOW_PATH" .json)
          TEST_NAME=$(echo "$FLOW_BASE" | tr '-' '_')
          TEST_FILE="$FLOW_DIR/tests/test_${TEST_NAME}.py"
          RESULT_FILE=$(mktemp --suffix=.xml)

          if [ -f "$TEST_FILE" ]; then
            pytest "$TEST_FILE" -v --tb=short --junitxml="$RESULT_FILE" || true
          else
            echo "No test file found at $TEST_FILE"
            echo "<testsuites><testsuite name=\"$FLOW_PATH\" tests=\"0\"/></testsuites>" > "$RESULT_FILE"
          fi
          echo "result_file=$RESULT_FILE" >> "$GITHUB_OUTPUT"
          if [ -f "$RESULT_FILE" ] && grep -q 'failures="[1-9]' "$RESULT_FILE"; then
            exit 1
          fi
          if [ -f "$RESULT_FILE" ] && grep -q 'errors="[1-9]' "$RESULT_FILE"; then
            exit 1
          fi

      - name: Teardown Runtime
        if: always() && steps.provision.outcome == 'success' && !contains(github.event.comment.body, 'do not clean')
        run: |
          python scripts/ci/teardown_ci_runtime.py \
            --config "${{ matrix.item.test_yaml }}" \
            --runtime-name "${{ steps.name.outputs.runtime_name }}" || true

      - name: Comment Result
        uses: actions/github-script@v9
        if: always()
        with:
          script: |
            const fs = require('fs');
            const flowPath = '${{ matrix.item.flow_path }}';
            const runtimeName = '${{ steps.name.outputs.runtime_name }}';
            const provisionOutcome = '${{ steps.provision.outcome }}';
            const deployOutcome = '${{ steps.deploy.outcome }}';
            const testsOutcome = '${{ steps.tests.outcome }}';
            const doNotClean = `${{ contains(github.event.comment.body, 'do not clean') }}` === 'true';
            const runUrl = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;

            let conclusion = 'success';
            let checkSummary = '';
            let body = '';

            if (provisionOutcome !== 'success') {
              body += `### ❌ Flow deployment failed\n\n`;
              body += `#### \`${flowPath}\`\n\n`;
              body += `**Runtime:** \`${runtimeName}\`\n\n`;
              body += `Runtime provisioning failed — see [workflow run](${runUrl})\n\n`;
              const provisionLog = '${{ steps.provision.outputs.error_log }}';
              if (provisionLog) {
                try {
                  const log = fs.readFileSync(provisionLog, 'utf8').trim();
                  if (log) {
                    const lines = log.split('\n');
                    const tail = lines.slice(-30).join('\n');
                    body += `<details><summary>Error output</summary>\n\n\`\`\`\n${tail}\n\`\`\`\n\n</details>\n\n`;
                  }
                } catch (e) {}
              }
              conclusion = 'failure';
              checkSummary = 'Runtime provisioning failed';
            } else if (deployOutcome !== 'success') {
              body += `### ❌ Flow deployment failed\n\n`;
              body += `#### \`${flowPath}\`\n\n`;
              body += `**Runtime:** \`${runtimeName}\`\n\n`;
              body += `Flow deployment failed — see [workflow run](${runUrl})\n\n`;
              const deployLog = '${{ steps.deploy.outputs.error_log }}';
              if (deployLog) {
                try {
                  const log = fs.readFileSync(deployLog, 'utf8').trim();
                  if (log) {
                    const lines = log.split('\n');
                    const tail = lines.slice(-30).join('\n');
                    body += `<details><summary>Error output</summary>\n\n\`\`\`\n${tail}\n\`\`\`\n\n</details>\n\n`;
                  }
                } catch (e) {}
              }
              conclusion = 'failure';
              checkSummary = 'Flow deployment failed';
            } else {
              const pgId = '${{ steps.deploy.outputs.pg_id }}';

              const resultFile = '${{ steps.tests.outputs.result_file }}';
              let testsPassed = 0;
              let testsFailed = 0;
              let testsErrors = 0;
              let testsSkipped = 0;
              let testcases = [];

              if (resultFile) {
                try {
                  const xml = fs.readFileSync(resultFile, 'utf8');
                  const suiteMatch = xml.match(/<testsuite[^>]*?\btests="(\d+)"[^>]*?\bfailures="(\d+)"[^>]*?\berrors="(\d+)"[^>]*?\bskipped="(\d+)"/);
                  if (suiteMatch) {
                    const total = parseInt(suiteMatch[1]);
                    testsFailed = parseInt(suiteMatch[2]);
                    testsErrors = parseInt(suiteMatch[3]);
                    testsSkipped = parseInt(suiteMatch[4]);
                    testsPassed = total - testsFailed - testsErrors - testsSkipped;
                  } else {
                    const fallback = xml.match(/<testsuite[^>]*?\btests="(\d+)"[^>]*?\bfailures="(\d+)"[^>]*?\berrors="(\d+)"/);
                    if (fallback) {
                      testsPassed = parseInt(fallback[1]) - parseInt(fallback[2]) - parseInt(fallback[3]);
                      testsFailed = parseInt(fallback[2]);
                      testsErrors = parseInt(fallback[3]);
                    }
                  }
                  testcases = [...xml.matchAll(/<testcase\s+[^>]*?classname="([^"]*)"[^>]*?name="([^"]*)"[^>]*?(?:\/>|>([\s\S]*?)<\/testcase>)/g)];
                } catch (e) {}
              }

              if (testsFailed + testsErrors > 0) {
                conclusion = 'failure';
                checkSummary = `${testsPassed} passed, ${testsFailed} failed, ${testsErrors} errors`;
                body += `### ❌ Flow deployment succeeded — tests failed\n\n`;
              } else {
                checkSummary = `${testsPassed} passed`;
                body += `### ✅ Flow deployment succeeded\n\n`;
              }

              body += `#### \`${flowPath}\`\n\n`;
              body += `**Process Group ID:** \`${pgId}\`\n\n`;

              if (testsOutcome === 'failure') {
                conclusion = 'failure';
                if (!checkSummary) checkSummary = 'Tests failed';
              }

              const total = testsPassed + testsFailed + testsErrors + testsSkipped;
              const icon = (testsFailed + testsErrors) === 0 ? '✅' : '❌';
              let summary = `${testsPassed} passed`;
              if (testsFailed > 0) summary += `, ${testsFailed} failed`;
              if (testsErrors > 0) summary += `, ${testsErrors} errors`;
              if (testsSkipped > 0) summary += `, ${testsSkipped} skipped`;

              body += `### Tests\n\n`;
              body += `${icon} **\`${flowPath}\`** — ${summary}\n\n`;

              if (testcases.length > 0) {
                body += `| Test | Result |\n`;
                body += `|------|--------|\n`;
                for (const tc of testcases) {
                  const testName = tc[2];
                  const inner = tc[3] || '';
                  if (inner.includes('<failure') || inner.includes('<error')) {
                    body += `| \`${testName}\` | ❌ failed |\n`;
                  } else if (inner.includes('<skipped')) {
                    body += `| \`${testName}\` | ⏭️ skipped |\n`;
                  } else {
                    body += `| \`${testName}\` | ✅ passed |\n`;
                  }
                }
                body += `\n`;

                const failures = testcases.filter(tc => (tc[3] || '').includes('<failure') || (tc[3] || '').includes('<error'));
                if (failures.length > 0) {
                  body += `### Failures\n\n`;
                  for (const tc of failures) {
                    const testName = tc[2];
                    const inner = tc[3] || '';
                    const msgMatch = inner.match(/message="([^"]*)"/);
                    const textMatch = inner.match(/<(?:failure|error)[^>]*>([\s\S]*?)<\/(?:failure|error)>/);
                    body += `#### \`${testName}\`\n\n`;
                    if (msgMatch) {
                      body += `> ${msgMatch[1].replace(/\n/g, '\n> ')}\n\n`;
                    }
                    if (textMatch) {
                      const text = textMatch[1].trim().substring(0, 500);
                      body += `\`\`\`\n${text}\n\`\`\`\n\n`;
                    }
                  }
                }
              }
            }

            if (doNotClean && provisionOutcome === 'success') {
              body += `### Cleanup\n\n`;
              body += `⏭️ Cleanup skipped — runtime \`${runtimeName}\` left running for debugging.\n\n`;
            } else if (provisionOutcome === 'success') {
              body += `### Cleanup\n\n`;
              body += `🧹 Cleaned up: \`${runtimeName}\`\n\n`;
            }

            body += `[Workflow run](${runUrl})`;

            await github.rest.issues.createComment({
              owner: context.repo.owner,
              repo: context.repo.repo,
              issue_number: context.issue.number,
              body
            });

            await github.rest.checks.update({
              owner: context.repo.owner,
              repo: context.repo.repo,
              check_run_id: ${{ steps.check.outputs.id }},
              status: 'completed',
              conclusion,
              completed_at: new Date().toISOString(),
              output: {
                title: conclusion === 'success' ? 'All tests passed' : 'Tests failed',
                summary: checkSummary || `Flow \`${flowPath}\` deploy ${conclusion}`
              }
            });