This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12021-25d6ba7f641b1f84bbf29281f1acb98bedc26a64 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 36b0c9fe8e89223b45e8a313d4e5a2ee59274575 Author: zhiweiniu <[email protected]> AuthorDate: Tue Sep 1 07:27:45 2026 +0000 [Feature][CI] Add benchmark profiling diagnostics (#12021) --- .github/workflows/backend.yml | 19 +- .github/workflows/benchmarks_diagnostics.yml | 253 +++++++++++++ docs/en/engines/zeta/benchmark.md | 87 +++++ docs/zh/engines/zeta/benchmark.md | 78 ++++ seatunnel-benchmarks/README.md | 108 +++--- tools/benchmarks/profile_benchmarks.sh | 338 +++++++++++++++++ tools/benchmarks/profile_report.py | 522 +++++++++++++++++++++++++++ tools/benchmarks/test_profile_benchmarks.py | 268 ++++++++++++++ tools/benchmarks/test_profile_report.py | 228 ++++++++++++ tools/benchmarks/test_regression_report.py | 148 ++++++++ tools/benchmarks/test_save_jmh_result.py | 180 +++++++++ 11 files changed, 2170 insertions(+), 59 deletions(-) diff --git a/.github/workflows/backend.yml b/.github/workflows/backend.yml index c641d9d3dc..371a404c7b 100644 --- a/.github/workflows/backend.yml +++ b/.github/workflows/backend.yml @@ -164,6 +164,7 @@ jobs: edge-agent: ${{ steps.filter.outputs.edge-agent }} edge-agent-e2e: ${{ steps.filter.outputs.edge-agent-e2e }} benchmarks: ${{ steps.filter.outputs.benchmarks }} + benchmark_tools: ${{ steps.filter.outputs.benchmark_tools }} docs: ${{ steps.filter.outputs.docs }} helm: ${{ steps.filter.outputs.helm }} ut-modules: ${{ steps.ut-modules.outputs.modules || '[]' }} @@ -226,11 +227,17 @@ jobs: echo "helm=$true_or_false" >> $GITHUB_OUTPUT echo "helm_files=$file_list" >> $GITHUB_OUTPUT - benchmark_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "seatunnel-benchmarks/**" "tools/benchmarks/**" ".github/workflows/benchmarks.yml"` + benchmark_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "seatunnel-benchmarks/**" "tools/benchmarks/**" ".github/workflows/benchmarks.yml" ".github/workflows/benchmarks_diagnostics.yml"` true_or_false=${benchmark_files%%$'\n'*} file_list=${benchmark_files#*$'\n'} echo "benchmarks=$true_or_false" >> $GITHUB_OUTPUT echo "benchmarks_files=$file_list" >> $GITHUB_OUTPUT + + benchmark_tool_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "tools/benchmarks/**"` + true_or_false=${benchmark_tool_files%%$'\n'*} + file_list=${benchmark_tool_files#*$'\n'} + echo "benchmark_tools=$true_or_false" >> $GITHUB_OUTPUT + echo "benchmark_tools_files=$file_list" >> $GITHUB_OUTPUT engine_e2e_files=`python tools/update_modules_check/check_file_updates.py ua $workspace apache/dev origin/$current_branch "seatunnel-e2e/seatunnel-engine-e2e/**"` true_or_false=${engine_e2e_files%%$'\n'*} @@ -562,6 +569,16 @@ jobs: -f 1 -wi 0 -i 1 -r 1s -foe true find "${result_dir}" -name '*.json' -print -quit | grep -q . + benchmark-tools-test: + needs: [ changes, sanity-check ] + if: needs.changes.outputs.benchmark_tools == 'true' + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + - name: Check benchmark tools + run: python3 -m unittest discover -s tools/benchmarks -p 'test_*.py' + updated-modules-integration-test-part-1: needs: [ changes, sanity-check ] if: needs.changes.outputs.api == 'false' && needs.changes.outputs.engine == 'false' && needs.changes.outputs.it-modules != '[]' diff --git a/.github/workflows/benchmarks_diagnostics.yml b/.github/workflows/benchmarks_diagnostics.yml new file mode 100644 index 0000000000..8b10adb5ce --- /dev/null +++ b/.github/workflows/benchmarks_diagnostics.yml @@ -0,0 +1,253 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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: Benchmarks Diagnostics + +on: + workflow_dispatch: + inputs: + seatunnel_ref: + description: 'SeaTunnel branch, tag, or commit SHA to diagnose.' + required: false + type: string + default: 'dev' + pr_number: + description: 'Optional trusted PR to diagnose. The workflow executes its code.' + required: false + type: string + benchmark: + description: 'One exact JMH benchmark method; multi-match selectors are rejected.' + required: true + type: string + default: 'IntermediateQueueBenchmark.disruptorRecordHandoff$' + java_version: + description: 'JDK used by the diagnostic run.' + required: true + type: choice + options: + - '8' + - '11' + default: '8' + profile: + description: 'Profiler mode. all runs CPU, wall-clock, lock, and GC separately.' + required: true + type: choice + options: + - 'cpu' + - 'wall' + - 'lock' + - 'gc' + - 'all' + default: 'cpu' + capture_jfr: + description: 'Also capture a JFR recording for offline JVM analysis.' + required: true + type: boolean + default: false + jmh_args: + description: 'Optional JMH args, for example: -wi 1 -i 1 -w 1s -r 1s. Forks are fixed at 1.' + required: false + type: string + +permissions: + contents: read + +jobs: + diagnostics: + name: Capture diagnostics with Java ${{ inputs.java_version }} + runs-on: ubuntu-24.04 + timeout-minutes: 240 + + env: + ASYNC_PROFILER_CPU_EVENT: 'ctimer' + ASYNC_PROFILER_VERSION: '4.5' + ASYNC_PROFILER_SHA256: '89546fbb9ee0fc5496c7edd4099b0709489bc78b0d8057ccbb4b801f6b032b62' + BENCHMARKS: ${{ inputs.benchmark }} + JAVA_VERSION: ${{ inputs.java_version }} + PROFILE_JMH_ARGS: ${{ inputs.jmh_args || '' }} + PR_NUMBER: ${{ inputs.pr_number || '' }} + SEATUNNEL_REF: ${{ inputs.seatunnel_ref || 'dev' }} + TARGET_DIRECTORY: target + steps: + - name: Checkout benchmark target + uses: actions/checkout@v4 + with: + fetch-depth: 0 + ref: ${{ inputs.pr_number != '' && format('refs/pull/{0}/head', inputs.pr_number) || inputs.seatunnel_ref || 'dev' }} + path: target + + - name: Checkout workflow tools + uses: actions/checkout@v4 + with: + ref: ${{ github.sha }} + path: workflow-source + sparse-checkout: tools/benchmarks + + - name: Set up JDK ${{ inputs.java_version }} with Maven cache + if: inputs.pr_number == '' + uses: actions/setup-java@v4 + with: + distribution: temurin + java-version: ${{ inputs.java_version }} + cache: maven + + - name: Set up JDK ${{ inputs.java_version }} without Maven cache + if: inputs.pr_number != '' + uses: actions/setup-java@v4 + with: + distribution: temurin + java-version: ${{ inputs.java_version }} + + - name: Build benchmark module + id: build_benchmark + working-directory: target + run: ./mvnw -Pbenchmark -pl seatunnel-benchmarks -am -DskipTests package + + - name: Install async-profiler + id: install_async_profiler + if: >- + steps.build_benchmark.outcome == 'success' && + (inputs.profile == 'cpu' || + inputs.profile == 'wall' || + inputs.profile == 'lock' || + inputs.profile == 'all') + run: | + archive="${RUNNER_TEMP}/async-profiler.tar.gz" + install_directory="${RUNNER_TEMP}/async-profiler" + curl --fail --location --retry 5 --retry-all-errors \ + --output "${archive}" \ + "https://github.com/async-profiler/async-profiler/releases/download/v${ASYNC_PROFILER_VERSION}/async-profiler-${ASYNC_PROFILER_VERSION}-linux-x64.tar.gz"; + echo "${ASYNC_PROFILER_SHA256} ${archive}" | sha256sum --check --strict + mkdir -p "${install_directory}" + tar --extract --gzip --file "${archive}" --directory "${install_directory}" --strip-components=1 + echo "ASYNC_PROFILER_HOME=${install_directory}" >> "${GITHUB_ENV}" + + - name: Profile CPU + if: >- + always() && + steps.build_benchmark.outcome == 'success' && + steps.install_async_profiler.outcome == 'success' && + (inputs.profile == 'cpu' || inputs.profile == 'all') + run: | + bash workflow-source/tools/benchmarks/profile_benchmarks.sh profile cpu \ + --repository "${TARGET_DIRECTORY}" \ + --output "${GITHUB_WORKSPACE}/benchmark-diagnostics/profile-cpu" + + - name: Profile wall clock + if: >- + always() && + steps.build_benchmark.outcome == 'success' && + steps.install_async_profiler.outcome == 'success' && + (inputs.profile == 'wall' || inputs.profile == 'all') + run: | + bash workflow-source/tools/benchmarks/profile_benchmarks.sh profile wall \ + --repository "${TARGET_DIRECTORY}" \ + --output "${GITHUB_WORKSPACE}/benchmark-diagnostics/profile-wall" + + - name: Profile locks + if: >- + always() && + steps.build_benchmark.outcome == 'success' && + steps.install_async_profiler.outcome == 'success' && + (inputs.profile == 'lock' || inputs.profile == 'all') + run: | + bash workflow-source/tools/benchmarks/profile_benchmarks.sh profile lock \ + --repository "${TARGET_DIRECTORY}" \ + --output "${GITHUB_WORKSPACE}/benchmark-diagnostics/profile-lock" + + - name: Profile GC and allocation + if: >- + always() && + steps.build_benchmark.outcome == 'success' && + (inputs.profile == 'gc' || inputs.profile == 'all') + run: | + bash workflow-source/tools/benchmarks/profile_benchmarks.sh profile gc \ + --repository "${TARGET_DIRECTORY}" \ + --output "${GITHUB_WORKSPACE}/benchmark-diagnostics/profile-gc" + + - name: Capture JFR + if: >- + always() && + steps.build_benchmark.outcome == 'success' && + inputs.capture_jfr + run: | + bash workflow-source/tools/benchmarks/profile_benchmarks.sh capture jfr \ + --repository "${TARGET_DIRECTORY}" \ + --output "${GITHUB_WORKSPACE}/benchmark-diagnostics/capture-jfr" + + - name: Upload CPU profile + if: always() && (inputs.profile == 'cpu' || inputs.profile == 'all') + uses: actions/upload-artifact@v4 + with: + name: seatunnel-benchmark-profile-cpu-java${{ env.JAVA_VERSION }}-${{ github.run_id }}-${{ github.run_attempt }} + retention-days: 30 + if-no-files-found: warn + path: benchmark-diagnostics/profile-cpu/** + + - name: Upload wall-clock profile + if: always() && (inputs.profile == 'wall' || inputs.profile == 'all') + uses: actions/upload-artifact@v4 + with: + name: seatunnel-benchmark-profile-wall-java${{ env.JAVA_VERSION }}-${{ github.run_id }}-${{ github.run_attempt }} + retention-days: 30 + if-no-files-found: warn + path: benchmark-diagnostics/profile-wall/** + + - name: Upload lock profile + if: always() && (inputs.profile == 'lock' || inputs.profile == 'all') + uses: actions/upload-artifact@v4 + with: + name: seatunnel-benchmark-profile-lock-java${{ env.JAVA_VERSION }}-${{ github.run_id }}-${{ github.run_attempt }} + retention-days: 30 + if-no-files-found: warn + path: benchmark-diagnostics/profile-lock/** + + - name: Upload GC profile + if: always() && (inputs.profile == 'gc' || inputs.profile == 'all') + uses: actions/upload-artifact@v4 + with: + name: seatunnel-benchmark-profile-gc-java${{ env.JAVA_VERSION }}-${{ github.run_id }}-${{ github.run_attempt }} + retention-days: 30 + if-no-files-found: warn + path: benchmark-diagnostics/profile-gc/** + + - name: Upload JFR capture + if: always() && inputs.capture_jfr + uses: actions/upload-artifact@v4 + with: + name: seatunnel-benchmark-capture-jfr-java${{ env.JAVA_VERSION }}-${{ github.run_id }}-${{ github.run_attempt }} + retention-days: 30 + if-no-files-found: warn + path: benchmark-diagnostics/capture-jfr/** + + - name: Publish diagnostic summary + if: always() + run: | + python3 workflow-source/tools/benchmarks/profile_report.py summary \ + --diagnostics-dir benchmark-diagnostics \ + --profile "${{ inputs.profile }}" \ + --capture-jfr "${{ inputs.capture_jfr }}" \ + --repository "${TARGET_DIRECTORY}" \ + --ref "${SEATUNNEL_REF}" \ + --pr-number "${PR_NUMBER}" \ + --benchmark "${BENCHMARKS}" \ + --java "${JAVA_VERSION}" \ + --jmh-args "${PROFILE_JMH_ARGS}" \ + --artifacts-url "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}#artifacts" \ + --run-id "${GITHUB_RUN_ID}" \ + --run-attempt "${GITHUB_RUN_ATTEMPT}" \ + >> "${GITHUB_STEP_SUMMARY}" diff --git a/docs/en/engines/zeta/benchmark.md b/docs/en/engines/zeta/benchmark.md index 6af8b54524..b20db41fa6 100644 --- a/docs/en/engines/zeta/benchmark.md +++ b/docs/en/engines/zeta/benchmark.md @@ -87,6 +87,13 @@ during each measured job instead of deferring file writes across several invocat ./mvnw -Pbenchmark -pl seatunnel-benchmarks -am -DskipTests package ``` +### Import the Module in IntelliJ IDEA + +The module is behind the inactive `benchmark` Maven profile, so IDEA may not import it when the +root project is first opened. In the Maven tool window, expand `Profiles`, enable `benchmark`, and +click `Reload All Maven Projects`. If the module is still absent, right-click +`seatunnel-benchmarks/pom.xml`, select `Add as Maven Project`, and reload Maven once more. + List every JMH method: ```bash @@ -142,6 +149,75 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelRowBenchmark \ For a quick functional validation, add `-f 1 -wi 0 -i 1 -r 1s` to shorten the run. A single un-warmed sample is not valid performance evidence. +### Read Workflow Reports + +The scheduled and manually triggered `Benchmarks` workflow runs each selected benchmark on Java 8 +and Java 11. Each Java job uploads one artifact containing: + +- the original `*.jmh.json`, preserving every fork and iteration sample; +- a versioned `*.report.json`, normalizing benchmark names, parameters, scores, errors, units, + direction, commit, JVM, CPU, and runner metadata; +- `summary.md`, which is also rendered in the GitHub Actions Job Summary; +- the environment fingerprint and any full-pipeline sample JSON. + +The normalized report includes median pipeline throughput, P50/P95/P99/max latency, latency growth, +completeness, and sustainable-sample counts. Raw samples and a schema version allow later tooling to +consume saved artifacts without parsing console logs. The workflow does not push results to a +repository branch. + +Manual runs offer common selectors through `benchmarks`; `custom_benchmarks` accepts a class, +method, or regular expression and overrides that choice. `.*` selects all current and future +benchmarks. When `pr_number` is set, the workflow executes `baseline -> PR -> PR -> baseline` on the +same worker, compares the median of both runs for each revision, and reports a direction-adjusted +percentage where a positive value is favorable. + +Absolute results remain sensitive to machine load, warmup, CPU frequency, and runner hardware. +Use GitHub-hosted results as trend and functional-check evidence. Prefer repeated baseline/change +runs on the same machine, as the PR comparison does, or use a fixed self-hosted runner for a future +regression gate. + +### Diagnose an Unstable Benchmark + +Use profiling only after a normal run shows an unexpected Score, Error, or CV. The diagnostic +runner keeps its report separate because profiler overhead makes its Score unsuitable for +regression comparisons. A diagnostic selector must resolve to exactly one benchmark method; +selectors such as `.*` or a class name that matches several methods are rejected. + +Install the complete async-profiler distribution and set `ASYNC_PROFILER_HOME` before running CPU, +wall-clock, or lock profiling. The runner records JFR first and uses the bundled `jfrconv` to create +forward and reverse flame graphs. GC profiling and JFR capture use JMH's built-in profilers and do +not need async-profiler: + +```bash +bash tools/benchmarks/profile_benchmarks.sh profile cpu --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +bash tools/benchmarks/profile_benchmarks.sh profile wall --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +bash tools/benchmarks/profile_benchmarks.sh profile lock --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +bash tools/benchmarks/profile_benchmarks.sh profile gc --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +bash tools/benchmarks/profile_benchmarks.sh capture jfr --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +``` + +CPU, wall-clock, and lock modes use JMH's async-profiler integration. GC mode uses JMH's GC +profiler, and `capture jfr` uses JMH's JFR profiler. The runner always uses exactly one fork so that +file-based profiler output cannot be overwritten by later forks. Warmup and measurement settings +still come from benchmark annotations unless they are overridden after `--`, for example +`-- -wi 1 -i 1 -w 1s -r 1s`. Each run gets a new default output directory; an explicit `--output` +directory must be empty to prevent stale artifacts from being mixed into the report. + +The raw JMH JSON records async-profiler as `secondaryMetrics.async` with a `NaN` Score because it +produces files rather than a numeric secondary metric. This is expected. The diagnostic report +shows the collected sample count; when lock profiling observes no contention, it reports zero +samples and intentionally omits an empty flame graph. + +The manual `Benchmarks Diagnostics` workflow requires one exact `benchmark` method and one +`java_version`. It is separate from the scheduled and manually triggered `Benchmarks` workflow, +which continues to run the Java 8/11 matrix. Selecting `all` runs CPU, wall-clock, lock, and GC +profiling as separate steps and uploads four independently downloadable artifacts; `capture_jfr` +adds a fifth JFR artifact. Each artifact contains only its mode's JFR recordings, flame graphs, text +summaries, JMH logs, and JSON reports. The job summary shows the target, benchmark settings, +per-mode results, and independent artifact names without repeating the full file inventory. On the +hosted Linux runner, CPU profiling uses async-profiler's `ctimer` event so it does not depend on +`perf_event` permissions. + ## Metrics ### Sample Validity @@ -223,6 +299,17 @@ Files under `pipeline-results` are custom JSON rather than JMH JSON. Inspect the `tools/benchmarks/save_jmh_result.py` and `tools/benchmarks/regression_report.py` to generate normalized JSON and Markdown reports. +## Add a Benchmark + +Keep cases small and focused on hot paths that run on one machine without external services. Useful +targets include `SeaTunnelRow` operations, format parsing and serialization, Transform hot paths, +connector option parsing, and split generation. + +Extend `BenchmarkBase` to inherit the shared JMH mode, forks, warmup, measurement, state, and output +unit defaults. Keep benchmark-specific state and setup in the benchmark class. Full-pipeline engine +lifecycle and controls belong in `SeaTunnelEnvironmentContext` or a focused subclass so checkpoint, +failure-recovery, and metrics scenarios can be added without duplicating cluster setup. + ## Performance Cost and Limitations - A pipeline benchmark starts an embedded Zeta cluster and requires at least 4 GiB of available heap. diff --git a/docs/zh/engines/zeta/benchmark.md b/docs/zh/engines/zeta/benchmark.md index f6f850b8fd..e4f91bc13d 100644 --- a/docs/zh/engines/zeta/benchmark.md +++ b/docs/zh/engines/zeta/benchmark.md @@ -82,6 +82,13 @@ Source 使用基于绝对时间的开环调度。每条记录都携带计划生 ./mvnw -Pbenchmark -pl seatunnel-benchmarks -am -DskipTests package ``` +### 在 IntelliJ IDEA 中导入模块 + +该模块位于默认未启用的 `benchmark` Maven profile 中，因此首次打开根项目时 IDEA 可能不会 +自动导入。在 Maven 工具窗口中展开 `Profiles`，启用 `benchmark`，然后点击 +`Reload All Maven Projects`。如果仍未显示该模块，右键点击 +`seatunnel-benchmarks/pom.xml`，选择 `Add as Maven Project`，再重新加载一次 Maven。 + 查看全部 JMH 方法： ```bash @@ -136,6 +143,67 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelRowBenchmark \ 快速功能验证时可以增加 `-f 1 -wi 0 -i 1 -r 1s` 缩短运行时间。没有预热且只有一个样本的 结果不能用于性能结论。 +### 查看 Workflow 报告 + +定时或手动触发的 `Benchmarks` workflow 会在 Java 8 和 Java 11 上运行所选 benchmark。 +每个 Java job 会上传一个 artifact，其中包含： + +- 原始 `*.jmh.json`，保留所有 fork 和 iteration 样本； +- 带版本的 `*.report.json`，统一记录 benchmark 名称、参数、Score、Error、单位、优化方向、 + Commit、JVM、CPU 和 Runner 元数据； +- `summary.md`，同时展示在 GitHub Actions Job Summary 中； +- 环境指纹和完整 Pipeline 的样本 JSON（如有）。 + +标准化报告还包含 Pipeline 吞吐中位数、P50/P95/P99/最大延迟、延迟增长、完整性和可持续样本 +数量。保留原始样本和 Schema 版本后，后续工具无需解析控制台日志即可消费已有 artifact。 +该 workflow 不会把结果推送到仓库分支。 + +手动运行可以通过 `benchmarks` 选择常用 selector；`custom_benchmarks` 可以填写类名、方法名或 +正则表达式，并覆盖前者。`.*` 会选择当前及未来的所有 benchmark。设置 `pr_number` 后， +workflow 会在同一 Worker 上按 `baseline -> PR -> PR -> baseline` 顺序运行，比较两个版本各自 +两次结果的中位数，并输出经过优化方向校正的百分比；正值表示 Candidate 向更优方向变化。 + +绝对结果仍会受到机器负载、预热、CPU 频率和 Runner 硬件影响。GitHub 托管 Runner 的结果 +适合用于趋势与功能检查。精确比较应像 PR 对比一样，在同一台机器上重复运行 Base 与 Change； +未来如需作为回归门禁，则应使用固定的 Self-hosted Runner。 + +### 诊断不稳定的 Benchmark + +只有正常运行出现异常的 Score、Error 或 CV 后，才使用 profiling 继续定位。Profiler 会引入 +额外开销，因此诊断报告与正常报告完全分开，诊断 Score 不能用于性能回归比较。诊断 selector +必须且只能匹配一个 benchmark 方法；`.*` 或能够匹配多个方法的类名会被拒绝。 + +运行 CPU、wall-clock 或 lock profiling 前，需要安装完整的 async-profiler，并设置 +`ASYNC_PROFILER_HOME`。Runner 会先生成 JFR，再使用安装包内的 `jfrconv` 生成正向和反向 +火焰图。GC profiling 和 JFR capture 使用 JMH 内置 profiler，不依赖 async-profiler： + +```bash +bash tools/benchmarks/profile_benchmarks.sh profile cpu --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +bash tools/benchmarks/profile_benchmarks.sh profile wall --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +bash tools/benchmarks/profile_benchmarks.sh profile lock --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +bash tools/benchmarks/profile_benchmarks.sh profile gc --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +bash tools/benchmarks/profile_benchmarks.sh capture jfr --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +``` + +CPU、wall-clock 和 lock 模式直接使用 JMH 的 async-profiler 集成，GC 模式使用 JMH GC +profiler，`capture jfr` 使用 JMH JFR profiler。Runner 始终使用一个 fork，防止后续 fork +覆盖文件型 profiler 的产物。预热和测量设置默认仍来自 benchmark 注解；如需覆盖，将参数 +放在 `--` 后，例如 `-- -wi 1 -i 1 -w 1s -r 1s`。默认输出目录每次运行都不同；显式指定 +的 `--output` 目录必须为空，避免旧产物混入报告。 + +async-profiler 产生文件而不是数值型 secondary metric，因此原始 JMH JSON 中的 +`secondaryMetrics.async` Score 为 `NaN`，这是预期行为。诊断报告会显示采集到的样本数； +lock profiling 没有观察到竞争时会报告 0 个样本，并且不会生成没有内容的火焰图。 + +手动触发 `Benchmarks Diagnostics` workflow 时，必须指定一个精确的 `benchmark` 方法和一个 +`java_version`。该 workflow 与定时或手动触发的 `Benchmarks` workflow 相互独立，后者继续 +运行 Java 8/11 matrix。选择 `all` 会分别执行 CPU、wall-clock、lock 和 GC step，并上传四个 +可以独立下载的 artifact；`capture_jfr` 会增加第五个 JFR artifact。每个 artifact 只包含对应 +模式的 JFR、火焰图、文本摘要、JMH 日志和 JSON 报告。Job Summary 会集中显示目标、 +benchmark 设置、各模式结果和独立 artifact 名称，不再重复完整文件清单。GitHub 托管的 +Linux runner 使用 async-profiler 的 `ctimer` 事件进行 CPU profiling，不依赖 `perf_event` +权限。 + ## 指标 ### 样本有效性 @@ -215,6 +283,16 @@ JMH Visualizer 会把参数值拼成标签。例如 `600000:4:256:64` 依次表 `tools/benchmarks/save_jmh_result.py` 和 `tools/benchmarks/regression_report.py` 生成 标准化 JSON 和 Markdown 报告。 +## 添加 Benchmark + +Benchmark 应保持小而专注，优先选择无需外部服务、可以在单机运行的热点路径，例如 +`SeaTunnelRow` 操作、格式解析与序列化、Transform 热点、Connector 参数解析和 Split 生成。 + +新 Benchmark 应继承 `BenchmarkBase`，复用统一的 JMH Mode、Fork、预热、测量、State 和输出 +单位配置；Benchmark 自身只保留场景相关的状态与 Setup。完整 Pipeline 的引擎生命周期和控制 +逻辑应放在 `SeaTunnelEnvironmentContext` 或职责明确的子类中，以便后续增加 Checkpoint、 +故障恢复和 Metrics 场景时无需复制集群 Setup。 + ## 开销与限制 - Pipeline Benchmark 会在本机启动嵌入式 Zeta 集群，需要至少 4 GiB 可用堆内存。 diff --git a/seatunnel-benchmarks/README.md b/seatunnel-benchmarks/README.md index 2204bbfc9f..506bdf419f 100644 --- a/seatunnel-benchmarks/README.md +++ b/seatunnel-benchmarks/README.md @@ -49,79 +49,71 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelRowBenchmark \ -rff seatunnel-benchmarks/target/benchmark-result.json ``` -## Benchmark reports - -The `Benchmarks` GitHub Actions workflow builds three artifacts for every Java version: +## Run Zeta full-pipeline benchmarks -- the original JMH JSON, which preserves all forks and iteration samples; -- a versioned `*.report.json`, which normalizes benchmark name, parameters, score, error, unit, - direction, commit, JVM, CPU, and runner image; -- a Markdown report, which is also displayed in the GitHub Actions job summary. +```bash +java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelPipelineBenchmark +``` -The normalized report also includes median pipeline throughput, latency P50/P95/P99/max, latency -growth, completeness, and the number of sustainable samples. Keeping the raw samples and a schema -version allows a later Codespeed service or regression checker to consume saved artifacts without -parsing console logs. The workflow also saves the CPU, kernel, runner image, memory, and JDK -fingerprint. It does not push benchmark data to a repository branch. +## Install async-profiler -Manual runs provide a `benchmarks` dropdown for common JMH selectors. The optional -`custom_benchmarks` input accepts any class name, method name, or regular expression and overrides -the dropdown, so a new benchmark can run before it is added to the common choices. `.*` runs every -current and future benchmark. An optional PR number compares that PR with `seatunnel_ref` on the -same worker in `base -> PR -> PR -> base` order. The comparison report uses the median of the two -baseline and two candidate runs and shows a direction-adjusted percentage; positive means the -candidate moved in the favorable direction. +CPU, wall-clock, and lock profiling require async-profiler's library and bundled `jfrconv`. +Set `ASYNC_PROFILER_HOME` to the complete installation. On macOS with Homebrew: -JMH treats selectors as regular expressions. Append `$` to an exact method selector when other -method names share the same prefix. +```bash +brew install async-profiler +export ASYNC_PROFILER_HOME="$(brew --prefix async-profiler)" +``` -GitHub-hosted runners can execute the workflow reliably while still having materially different -host CPU performance. Treat these artifacts as trend and functional-check data, not as a regression -gate based on one run or on JMH's within-run `scoreError`. A future regression gate should compare -the base and change on the same worker or use a fixed self-hosted runner. +On Linux x64: -## IntelliJ IDEA +```bash +SEATUNNEL_ASYNC_PROFILER_VERSION=4.5 +SEATUNNEL_ASYNC_PROFILER_HOME="${PWD}/seatunnel-benchmarks/target/async-profiler" +SEATUNNEL_ASYNC_PROFILER_ARCHIVE="${SEATUNNEL_ASYNC_PROFILER_HOME}.tar.gz" + +curl --fail --location --retry 5 --retry-all-errors \ + --output "${SEATUNNEL_ASYNC_PROFILER_ARCHIVE}" \ + "https://github.com/async-profiler/async-profiler/releases/download/v${SEATUNNEL_ASYNC_PROFILER_VERSION}/async-profiler-${SEATUNNEL_ASYNC_PROFILER_VERSION}-linux-x64.tar.gz"; +echo "89546fbb9ee0fc5496c7edd4099b0709489bc78b0d8057ccbb4b801f6b032b62 ${SEATUNNEL_ASYNC_PROFILER_ARCHIVE}" \ + | sha256sum --check --strict +mkdir -p "${SEATUNNEL_ASYNC_PROFILER_HOME}" +tar --extract --gzip \ + --file "${SEATUNNEL_ASYNC_PROFILER_ARCHIVE}" \ + --directory "${SEATUNNEL_ASYNC_PROFILER_HOME}" \ + --strip-components=1 +export ASYNC_PROFILER_HOME="${SEATUNNEL_ASYNC_PROFILER_HOME}" +``` -The benchmark module is behind the inactive `benchmark` Maven profile, so IDEA may not import it -automatically after opening the SeaTunnel root project. +## Profile one benchmark -To make IDEA recognize the module: +```bash +bash tools/benchmarks/profile_benchmarks.sh profile cpu \ + --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' -1. Open the Maven tool window. -2. Expand `Profiles`. -3. Enable the `benchmark` profile. -4. Click `Reload All Maven Projects`. +bash tools/benchmarks/profile_benchmarks.sh profile wall \ + --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' -If the module is still not shown, right-click `seatunnel-benchmarks/pom.xml` and choose -`Add as Maven Project`, then reload Maven once more. +bash tools/benchmarks/profile_benchmarks.sh profile lock \ + --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' -## Interpreting results +bash tools/benchmarks/profile_benchmarks.sh profile gc \ + --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' -Benchmark results are sensitive to machine load, JVM warmup, CPU frequency, and runner type. Prefer -comparing repeated baseline/change runs on the same machine instead of comparing absolute numbers -from different machines. +bash tools/benchmarks/profile_benchmarks.sh capture jfr \ + --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +``` -## Run Zeta full-pipeline benchmarks +GC and JFR use JMH's built-in profilers and do not require async-profiler. Override JMH warmup and +measurement settings after `--`: ```bash -java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelPipelineBenchmark +bash tools/benchmarks/profile_benchmarks.sh profile gc \ + --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' \ + -- -wi 1 -i 1 -w 1s -r 1s ``` -See the [Zeta benchmark guide](../docs/en/engines/zeta/benchmark.md) for architecture, parameters, -resource settings, and result interpretation. - -## Adding benchmarks - -Keep benchmark cases small and focused. Good first targets are hot paths that can run on a single -machine without external services, such as: - -- `SeaTunnelRow` operations -- format parsing and serialization -- transform hot paths -- connector option parsing -- split generation logic +The profiling script accepts exactly one benchmark method and always uses one fork. -Common JMH and JVM settings live in the single `BenchmarkBase`. Individual benchmark classes -extend it and only define their own data setup and benchmark methods. Engine lifecycle and -engine-level controls belong in `SeaTunnelEnvironmentContext`, so checkpoint, failure-recovery, -and metrics scenarios can be added without duplicating cluster setup in each benchmark. +See the [Zeta benchmark guide](../docs/en/engines/zeta/benchmark.md) for workflow reports, IntelliJ +IDEA setup, benchmark parameters, metrics, result interpretation, and contribution guidance. diff --git a/tools/benchmarks/profile_benchmarks.sh b/tools/benchmarks/profile_benchmarks.sh new file mode 100644 index 0000000000..0a54eb0d51 --- /dev/null +++ b/tools/benchmarks/profile_benchmarks.sh @@ -0,0 +1,338 @@ +#!/usr/bin/env bash +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. +# + +set -euo pipefail + +script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +default_repository_dir=$(cd "${script_dir}/../.." && pwd) + +usage() { + cat <<'EOF' +Usage: + profile_benchmarks.sh profile <cpu|wall|lock|gc> [options] [-- JMH_ARGS...] + profile_benchmarks.sh capture jfr [options] [-- JMH_ARGS...] + +Options: + --benchmark REGEX JMH selector. Defaults to $BENCHMARKS or one queue benchmark method. + The selector must resolve to exactly one benchmark method. + --repository DIR Repository containing the benchmark JAR. Defaults to this script's repository. + --output DIR Output directory. Defaults below seatunnel-benchmarks/target/profiles. + -h, --help Show this help. + +Examples: + profile_benchmarks.sh profile cpu + profile_benchmarks.sh profile lock --benchmark 'IntermediateQueueBenchmark.blockingQueueRecordHandoff$' + profile_benchmarks.sh profile gc -- -f 1 -wi 1 -i 1 -w 1s -r 1s + profile_benchmarks.sh capture jfr --benchmark 'IntermediateQueueBenchmark.disruptorRecordHandoff$' +EOF +} + +fail() { + echo "ERROR: $*" >&2 + exit 1 +} + +if [[ $# -eq 1 && ( "$1" == "-h" || "$1" == "--help" ) ]]; then + usage + exit 0 +fi + +if [[ $# -lt 2 ]]; then + usage >&2 + exit 1 +fi + +command_name="$1" +mode="$2" +shift 2 + +case "${command_name}" in + profile) + case "${mode}" in + cpu | wall | lock | gc) ;; + *) fail "Unsupported profile mode '${mode}'. Expected cpu, wall, lock, or gc." ;; + esac + ;; + capture) + case "${mode}" in + jfr) ;; + *) fail "Unsupported capture mode '${mode}'. Expected jfr." ;; + esac + ;; + -h | --help) + usage + exit 0 + ;; + *) fail "Unsupported command '${command_name}'. Expected profile or capture." ;; +esac + +benchmark_selector="${BENCHMARKS:-IntermediateQueueBenchmark.disruptorRecordHandoff$}" +repository_dir="${default_repository_dir}" +output_directory="" +jmh_arguments=() + +while [[ $# -gt 0 ]]; do + case "$1" in + --benchmark) + [[ $# -ge 2 ]] || fail "--benchmark requires a value" + benchmark_selector="$2" + shift 2 + ;; + --output) + [[ $# -ge 2 ]] || fail "--output requires a value" + output_directory="$2" + shift 2 + ;; + --repository) + [[ $# -ge 2 ]] || fail "--repository requires a value" + repository_dir="$2" + shift 2 + ;; + --) + shift + jmh_arguments=("$@") + break + ;; + -h | --help) + usage + exit 0 + ;; + *) fail "Unknown script option '$1'. Put JMH arguments after --." ;; + esac +done + +if [[ ${#jmh_arguments[@]} -eq 0 && -n "${PROFILE_JMH_ARGS:-}" ]]; then + read -r -a jmh_arguments <<< "${PROFILE_JMH_ARGS}" +fi + +if [[ "${repository_dir}" != /* ]]; then + repository_dir="${PWD}/${repository_dir}" +fi +[[ -d "${repository_dir}" ]] || fail "Repository directory does not exist: ${repository_dir}" +repository_dir=$(cd "${repository_dir}" && pwd) +benchmark_jar="${repository_dir}/seatunnel-benchmarks/target/benchmarks.jar" +[[ -f "${benchmark_jar}" ]] || fail "Missing ${benchmark_jar}. Build it with: ./mvnw -Pbenchmark -pl seatunnel-benchmarks -am -DskipTests package" + +fork_option_seen=false +for ((argument_index = 0; argument_index < ${#jmh_arguments[@]}; argument_index++)); do + case "${jmh_arguments[argument_index]}" in + -f) + [[ "${fork_option_seen}" == "false" ]] || fail "Specify the JMH -f option only once." + [[ $((argument_index + 1)) -lt ${#jmh_arguments[@]} ]] || fail "JMH -f requires a value." + [[ "${jmh_arguments[argument_index + 1]}" == "1" ]] || fail "Diagnostic runs require exactly one fork; use '-f 1'." + fork_option_seen=true + ;; + -f=*) + [[ "${fork_option_seen}" == "false" ]] || fail "Specify the JMH -f option only once." + [[ "${jmh_arguments[argument_index]}" == "-f=1" ]] || fail "Diagnostic runs require exactly one fork; use '-f 1'." + fork_option_seen=true + ;; + esac +done +if [[ "${fork_option_seen}" == "false" ]]; then + jmh_arguments+=( -f 1 ) +fi + +set +e +benchmark_listing=$(java -jar "${benchmark_jar}" "${benchmark_selector}" -l) +listing_status=$? +set -e +[[ ${listing_status} -eq 0 ]] || fail "Could not resolve the benchmark selector '${benchmark_selector}'." + +resolved_benchmarks=() +while IFS= read -r benchmark_name; do + case "${benchmark_name}" in + "" | Benchmarks:*) continue ;; + *) resolved_benchmarks+=( "${benchmark_name}" ) ;; + esac +done <<< "${benchmark_listing}" + +if [[ ${#resolved_benchmarks[@]} -eq 0 ]]; then + fail "Benchmark selector '${benchmark_selector}' did not match a benchmark." +fi +if [[ ${#resolved_benchmarks[@]} -ne 1 ]]; then + printf 'Matched benchmarks:\n' >&2 + printf ' %s\n' "${resolved_benchmarks[@]}" >&2 + fail "Benchmark diagnostics require one exact benchmark method; '${benchmark_selector}' matched ${#resolved_benchmarks[@]}." +fi + +if [[ -z "${output_directory}" ]]; then + run_id="$(date -u +%Y%m%dT%H%M%SZ)-$$" + output_directory="${repository_dir}/seatunnel-benchmarks/target/profiles/${command_name}-${mode}-${run_id}" +elif [[ "${output_directory}" != /* ]]; then + output_directory="${PWD}/${output_directory}" +fi + +if [[ -d "${output_directory}" ]]; then + first_existing_artifact=$(find "${output_directory}" -mindepth 1 -print -quit) + [[ -z "${first_existing_artifact}" ]] || fail "Output directory is not empty: ${output_directory}" +elif [[ -e "${output_directory}" ]]; then + fail "Output path is not a directory: ${output_directory}" +fi + +raw_directory="${output_directory}/raw" +pipeline_directory="${output_directory}/pipeline-results" +jmh_result="${output_directory}/result.jmh.json" +jmh_log="${output_directory}/jmh.log" +profile_json="${output_directory}/profile-report.json" +profile_markdown="${output_directory}/profile-summary.md" +mkdir -p "${raw_directory}" "${pipeline_directory}" + +resolve_async_profiler_library() { + local candidate + for candidate in \ + "${ASYNC_PROFILER_HOME}/lib/libasyncProfiler.so" \ + "${ASYNC_PROFILER_HOME}/lib/libasyncProfiler.dylib"; do + if [[ -f "${candidate}" ]]; then + echo "${candidate}" + return + fi + done + fail "async-profiler library was not found under ASYNC_PROFILER_HOME." +} + +convert_async_profiler_jfr() { + local profile_jfr + local profile_summary + profile_jfr=$(find "${raw_directory}" -type f -name 'jfr-*.jfr' -print -quit) + profile_summary=$(find "${raw_directory}" -type f -name 'summary-*.txt' -print -quit) + [[ -n "${profile_jfr}" && -n "${profile_summary}" ]] || return 1 + + local sample_count + sample_count=$( + awk -F ':' '/^Total samples/ {gsub(/[[:space:]]/, "", $2); print $2; exit}' \ + "${profile_summary}" + ) + if [[ ! "${sample_count}" =~ ^[0-9]+$ ]]; then + echo "ERROR: Could not read async-profiler sample count from ${profile_summary}." >&2 + return 1 + fi + if [[ ${sample_count} -eq 0 ]]; then + echo "async-profiler collected no ${mode} samples; no flame graph was generated." + return 0 + fi + + local event_option="--${mode}" + local profile_prefix + profile_prefix=$(basename "${profile_jfr}" .jfr) + profile_prefix=${profile_prefix#jfr-} + local profile_directory + profile_directory=$(dirname "${profile_jfr}") + local forward_html="${profile_directory}/flame-${profile_prefix}-forward.html" + local reverse_html="${profile_directory}/flame-${profile_prefix}-reverse.html" + + "${async_profiler_converter}" --output html "${event_option}" --threads \ + "${profile_jfr}" "${forward_html}" + "${async_profiler_converter}" --output html "${event_option}" --threads --reverse \ + "${profile_jfr}" "${reverse_html}" +} + +profiler_arguments=() +async_profiler_converter="" +case "${command_name}:${mode}" in + profile:cpu | profile:wall | profile:lock) + [[ -n "${ASYNC_PROFILER_HOME:-}" ]] || fail "Set ASYNC_PROFILER_HOME to the complete async-profiler installation." + async_profiler_library=$(resolve_async_profiler_library) + async_profiler_converter="${ASYNC_PROFILER_HOME}/bin/jfrconv" + [[ -x "${async_profiler_converter}" ]] || fail "Missing async-profiler converter: ${async_profiler_converter}" + event="${mode}" + interval=10000000 + if [[ "${mode}" == "cpu" ]]; then + event="${ASYNC_PROFILER_CPU_EVENT:-cpu}" + fi + if [[ "${mode}" == "lock" ]]; then + # Lock profiling uses the interval as a contention-duration threshold. A low threshold + # keeps short queue monitor contention visible; profiler scores remain diagnostic only. + interval=10000 + fi + profiler_arguments+=( + -prof + "async:libPath=${async_profiler_library};event=${event};interval=${interval};threads=true;output=jfr,text;dir=${raw_directory}" + ) + ;; + profile:gc) + profiler_arguments+=( + -prof + "gc:alloc=true;churn=true;churnWait=500" + ) + ;; + capture:jfr) + profiler_arguments+=( + -prof + "jfr:dir=${raw_directory};configName=profile;stackDepth=256" + ) + ;; +esac + +java_command=( + java + "-Dseatunnel.benchmark.result.dir=${pipeline_directory}" + -jar + "${benchmark_jar}" + "${benchmark_selector}" + -foe + true + -rf + json + -rff + "${jmh_result}" +) +java_command+=("${profiler_arguments[@]}") +java_command+=("${jmh_arguments[@]}") + +echo "Diagnostic: ${command_name} ${mode}" +echo "Benchmark: ${resolved_benchmarks[0]}" +echo "Output: ${output_directory}" +echo "JMH log: ${jmh_log}" + +set +e +( + cd "${repository_dir}" + "${java_command[@]}" +) > "${jmh_log}" 2>&1 +benchmark_status=$? +set -e + +diagnostic_status=${benchmark_status} +if [[ ${benchmark_status} -eq 0 && -n "${async_profiler_converter}" ]]; then + if ! convert_async_profiler_jfr; then + diagnostic_status=1 + fi +fi + +report_status=0 +python3 "${script_dir}/profile_report.py" mode \ + --mode "${mode}" \ + --command "${command_name}" \ + --jmh "${jmh_result}" \ + --artifact-dir "${output_directory}" \ + --output-json "${profile_json}" \ + --output-md "${profile_markdown}" \ + --status "${diagnostic_status}" || report_status=$? + +if [[ ${benchmark_status} -ne 0 ]]; then + echo "JMH failed with status ${benchmark_status}; the last 200 log lines follow:" >&2 + tail -n 200 "${jmh_log}" >&2 + exit "${benchmark_status}" +fi +if [[ ${diagnostic_status} -ne 0 ]]; then + echo "async-profiler post-processing failed; inspect ${profile_markdown}." >&2 + exit "${diagnostic_status}" +fi +echo "Diagnostic report: ${profile_markdown}" +exit "${report_status}" diff --git a/tools/benchmarks/profile_report.py b/tools/benchmarks/profile_report.py new file mode 100644 index 0000000000..d091990d02 --- /dev/null +++ b/tools/benchmarks/profile_report.py @@ -0,0 +1,522 @@ +#!/usr/bin/env python3 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. +# + +"""Build a standalone report for benchmark profiling artifacts.""" + +import argparse +import datetime +import json +import math +import pathlib +import re +import subprocess + + +SCHEMA = "seatunnel-profile/v1" +GC_COLUMNS = ( + ("gc.alloc.rate.norm", "Alloc/op", "B/op"), + ("gc.alloc.rate", "Alloc rate", "MB/sec"), + ("gc.count", "GC count", "counts"), + ("gc.time", "GC time", "ms"), +) +PROFILE_MODES = ("cpu", "wall", "lock", "gc") +MODE_DETAILS = { + "cpu": { + "directory": "profile-cpu", + "label": "CPU", + "contents": "Flame graphs, async-profiler JFR and text summary", + }, + "wall": { + "directory": "profile-wall", + "label": "Wall", + "contents": "Flame graphs, async-profiler JFR and text summary", + }, + "lock": { + "directory": "profile-lock", + "label": "Lock", + "contents": "Contention flame graphs, JFR and text summary", + }, + "gc": { + "directory": "profile-gc", + "label": "GC", + "contents": "Allocation and garbage-collection metrics", + }, + "jfr": { + "directory": "capture-jfr", + "label": "JFR", + "contents": "JVM Flight Recorder capture", + }, +} + + +def parse_args(): + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="report_command", required=True) + + mode_parser = commands.add_parser("mode", help="Build one profiler report.") + mode_parser.add_argument("--mode", required=True) + mode_parser.add_argument( + "--command", required=True, choices=("profile", "capture") + ) + mode_parser.add_argument("--jmh", required=True, type=pathlib.Path) + mode_parser.add_argument("--artifact-dir", required=True, type=pathlib.Path) + mode_parser.add_argument("--output-json", required=True, type=pathlib.Path) + mode_parser.add_argument("--output-md", required=True, type=pathlib.Path) + mode_parser.add_argument("--status", required=True, type=int) + + summary_parser = commands.add_parser( + "summary", help="Build the combined GitHub Actions summary." + ) + summary_parser.add_argument( + "--diagnostics-dir", required=True, type=pathlib.Path + ) + summary_parser.add_argument( + "--profile", required=True, choices=("none", "cpu", "wall", "lock", "gc", "all") + ) + summary_parser.add_argument( + "--capture-jfr", required=True, choices=("true", "false") + ) + summary_parser.add_argument("--repository", required=True, type=pathlib.Path) + summary_parser.add_argument("--ref", required=True) + summary_parser.add_argument("--pr-number", default="") + summary_parser.add_argument("--benchmark", required=True) + summary_parser.add_argument("--java", required=True) + summary_parser.add_argument("--jmh-args", default="") + summary_parser.add_argument("--artifacts-url", required=True) + summary_parser.add_argument("--run-id", required=True) + summary_parser.add_argument("--run-attempt", required=True) + return parser.parse_args() + + +def finite_or_none(value): + if value is None: + return None + number = float(value) + return number if math.isfinite(number) else None + + +def load_jmh(path): + if not path.is_file(): + return [] + try: + with path.open(encoding="utf-8") as handle: + return json.load(handle) + except (OSError, json.JSONDecodeError): + return [] + + +def load_profile_report(path): + if not path.is_file(): + return None + try: + with path.open(encoding="utf-8") as handle: + report = json.load(handle) + except (OSError, json.JSONDecodeError): + return None + return report if isinstance(report, dict) else None + + +def resolve_commit(repository): + try: + result = subprocess.run( + ["git", "-C", str(repository), "rev-parse", "--short=12", "HEAD"], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + ) + except (OSError, subprocess.CalledProcessError): + return "unknown" + return result.stdout.strip() or "unknown" + + +def benchmark_name(result): + return result.get("benchmark", "unknown") + + +def compact_params(params): + if not params: + return "default" + return ", ".join("{}={}".format(key, params[key]) for key in sorted(params)) + + +def gc_metrics(results): + metrics = [] + for result in results: + secondary = result.get("secondaryMetrics", {}) + values = {} + for name, metric in secondary.items(): + if not name.startswith("gc."): + continue + values[name] = { + "value": finite_or_none(metric.get("score")), + "error": finite_or_none(metric.get("scoreError")), + "unit": metric.get("scoreUnit"), + } + if values: + metrics.append( + { + "benchmark": benchmark_name(result), + "params": result.get("params", {}), + "metrics": values, + } + ) + return metrics + + +def human_size(size): + value = float(size) + for unit in ("B", "KiB", "MiB", "GiB", "TiB"): + if abs(value) < 1024.0 or unit == "TiB": + return "{:.1f} {}".format(value, unit) + value /= 1024.0 + return "{} B".format(size) + + +def artifacts(directory, excluded): + values = [] + if not directory.is_dir(): + return values + excluded_paths = {path.resolve() for path in excluded} + for path in sorted(directory.rglob("*")): + if not path.is_file() or path.resolve() in excluded_paths: + continue + values.append( + { + "path": str(path.relative_to(directory)), + "size_bytes": path.stat().st_size, + } + ) + return values + + +def async_sample_count(directory): + counts = [] + if directory.is_dir(): + for summary in directory.rglob("summary-*.txt"): + match = re.search( + r"^Total samples\s*:\s*(\d+)", + summary.read_text(encoding="utf-8"), + flags=re.MULTILINE, + ) + if match: + counts.append(int(match.group(1))) + return sum(counts) if counts else None + + +def format_number(value): + if value is None: + return "n/a" + if abs(value) >= 1000: + return "{:,.2f}".format(value) + if abs(value) >= 1: + return "{:.3f}".format(value) + return "{:.6f}".format(value) + + +def gc_report_lines(metrics): + """Return GC/allocation Markdown lines, or no section when metrics are unavailable.""" + if not metrics: + return [] + lines = [ + "### GC and allocation metrics", + "", + "These values come from JMH's GC profiler and cover measurement iterations only. " + "The primary JMH Score from a profiled run is diagnostic and is not compared with " + "normal benchmark results.", + "", + "- **Alloc/op (B/op):** average bytes allocated per benchmark operation. This is the " + "most direct allocation-efficiency metric; lower is generally better.", + "- **Alloc rate (MB/sec):** average allocation throughput. It depends on both bytes " + "allocated per operation and operations per second, so a high value alone does not " + "prove a regression.", + "- **GC count (counts):** total garbage collections observed across the measured " + "iterations and forks. It is not normalized per operation.", + "- **GC time (ms):** accumulated collection time reported by the JVM across the " + "measured iterations and forks. It is JVM- and collector-dependent and is not an " + "exact stop-the-world pause-time measurement.", + "", + "Compare these values only when the JVM, collector, heap settings, benchmark selector, " + "and JMH fork/warmup/measurement settings are the same. Capture JFR separately when " + "collection phases and pause details are needed.", + "", + "| Benchmark | Parameters | Alloc/op | Alloc rate | GC count | GC time |", + "| --- | --- | ---: | ---: | ---: | ---: |", + ] + for result in metrics: + row = [] + for name, _, unit in GC_COLUMNS: + metric = result["metrics"].get(name) + row.append( + "{} {}".format(format_number(metric["value"]), unit) if metric else "n/a" + ) + lines.append( + "| `{}` | `{}` | {} | {} | {} | {} |".format( + result["benchmark"].rsplit(".", 1)[-1], + compact_params(result["params"]), + *row, + ) + ) + return lines + + +def artifact_report_lines(values): + if not values: + return ["### Artifacts", "", "No diagnostic artifacts were produced."] + lines = [ + "### Artifacts", + "", + "Open these files locally, or download the workflow artifact for offline analysis.", + "", + "| File | Size |", + "| --- | ---: |", + ] + for artifact in values: + lines.append( + "| `{}` | {} |".format( + artifact["path"], human_size(artifact["size_bytes"]) + ) + ) + return lines + + +def report_status(report): + if report is None: + return "not produced" + status = report.get("status") + if status == 0: + return "passed" + return "failed ({})".format(status if status is not None else "unknown") + + +def build_report(args): + jmh_results = load_jmh(args.jmh) + gc = gc_metrics(jmh_results) + generated_artifacts = artifacts( + args.artifact_dir, (args.output_json, args.output_md) + ) + return { + "schema": SCHEMA, + "generated_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "command": args.command, + "mode": args.mode, + "status": args.status, + "score_comparable": False, + "async_samples": async_sample_count(args.artifact_dir), + "gc_metrics": gc, + "artifacts": generated_artifacts, + } + + +def render_markdown(report): + lines = [ + "## {} {} diagnostics".format( + report["mode"].upper(), report["command"] + ), + "", + "- Status: `{}`".format(report_status(report)), + "- Schema: `{}`".format(report["schema"]), + "- Score comparable with normal benchmarks: `no`", + ] + if report["mode"] in ("cpu", "wall", "lock"): + sample_count = report.get("async_samples") + lines.extend( + [ + "- Async-profiler samples: `{}`".format( + sample_count if sample_count is not None else "unknown" + ), + "- `secondaryMetrics.async` in the raw JMH JSON is a file-profiler marker; " + "its `NaN` Score is expected and is not a missing performance metric.", + ] + ) + if sample_count == 0: + lines.extend( + [ + "", + "> No matching {} events were observed, so no flame graph was generated.".format( + report["mode"] + ), + ] + ) + for section in ( + gc_report_lines(report["gc_metrics"]), + artifact_report_lines(report["artifacts"]), + ): + if section: + lines.extend([""] + section) + return "\n".join(lines) + "\n" + + +def selected_modes(profile, capture_jfr): + modes = [] + if profile == "all": + modes.extend(PROFILE_MODES) + elif profile != "none": + modes.append(profile) + if capture_jfr: + modes.append("jfr") + return modes + + +def mode_report(root, mode): + directory = root / MODE_DETAILS[mode]["directory"] + return load_profile_report(directory / "profile-report.json") + + +def workflow_mode_lines(mode, report): + details = MODE_DETAILS[mode] + lines = [ + "### {} diagnostics".format(details["label"]), + "", + "- Status: `{}`".format(report_status(report)), + ] + if report is None: + return lines + if mode in ("cpu", "wall", "lock"): + sample_count = report.get("async_samples") + lines.append( + "- Async-profiler samples: `{}`".format( + sample_count if sample_count is not None else "unknown" + ) + ) + if sample_count == 0: + lines.extend( + [ + "", + "> No matching {} events were observed, so no flame graph was generated.".format( + mode + ), + ] + ) + if mode == "gc": + gc_lines = gc_report_lines(report.get("gc_metrics", [])) + if gc_lines: + gc_lines[0] = "#### GC and allocation metrics" + lines.extend([""] + gc_lines) + return lines + + +def workflow_artifact_name(mode, java, run_id, run_attempt): + if mode == "jfr": + prefix = "seatunnel-benchmark-capture-jfr" + else: + prefix = "seatunnel-benchmark-profile-{}".format(mode) + return "{}-java{}-{}-{}".format(prefix, java, run_id, run_attempt) + + +def render_workflow_summary( + diagnostics_dir, + profile, + capture_jfr, + target_ref, + pr_number, + commit, + benchmark, + java, + jmh_args, + artifacts_url, + run_id, + run_attempt, +): + """Return the combined GitHub Actions summary for the selected diagnostic modes.""" + modes = selected_modes(profile, capture_jfr) + reports = {mode: mode_report(diagnostics_dir, mode) for mode in modes} + target = "PR #{}".format(pr_number) if pr_number else target_ref + settings = jmh_args or "benchmark annotations" + lines = [ + "## Benchmark diagnostics", + "", + "- Target: `{}` at `{}`".format(target, commit), + "- Benchmark: `{}`".format(benchmark), + "- Java: `{}`".format(java), + "- JMH settings: `{}`".format(settings), + "- Profiled scores comparable with normal benchmarks: `no`", + ] + if any(mode in ("cpu", "wall", "lock") for mode in modes): + lines.extend( + [ + "- `secondaryMetrics.async` is a file-profiler marker; its `NaN` Score " + "is expected.", + ] + ) + for mode in modes: + lines.extend([""] + workflow_mode_lines(mode, reports[mode])) + + lines.extend( + [ + "", + "### Downloads", + "", + "Download each selected mode independently from the " + "[run artifacts]({}).".format(artifacts_url), + "", + "| Mode | Status | Artifact | Contents |", + "| --- | --- | --- | --- |", + ] + ) + for mode in modes: + details = MODE_DETAILS[mode] + directory = diagnostics_dir / details["directory"] + artifact = ( + workflow_artifact_name(mode, java, run_id, run_attempt) + if directory.is_dir() + else "not produced" + ) + lines.append( + "| {} | `{}` | `{}` | {} |".format( + details["label"], + report_status(reports[mode]), + artifact, + details["contents"], + ) + ) + return "\n".join(lines) + "\n" + + +def main(): + args = parse_args() + if args.report_command == "summary": + print( + render_workflow_summary( + diagnostics_dir=args.diagnostics_dir, + profile=args.profile, + capture_jfr=args.capture_jfr == "true", + target_ref=args.ref, + pr_number=args.pr_number, + commit=resolve_commit(args.repository), + benchmark=args.benchmark, + java=args.java, + jmh_args=args.jmh_args, + artifacts_url=args.artifacts_url, + run_id=args.run_id, + run_attempt=args.run_attempt, + ), + end="", + ) + return + + report = build_report(args) + args.output_json.parent.mkdir(parents=True, exist_ok=True) + with args.output_json.open("w", encoding="utf-8") as handle: + json.dump(report, handle, indent=2, sort_keys=True) + handle.write("\n") + args.output_md.write_text(render_markdown(report), encoding="utf-8") + + +if __name__ == "__main__": + main() diff --git a/tools/benchmarks/test_profile_benchmarks.py b/tools/benchmarks/test_profile_benchmarks.py new file mode 100644 index 0000000000..1eab3f3a67 --- /dev/null +++ b/tools/benchmarks/test_profile_benchmarks.py @@ -0,0 +1,268 @@ +#!/usr/bin/env python3 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. +# + +import os +import pathlib +import subprocess +import tempfile +import unittest + + +SCRIPT = pathlib.Path(__file__).with_name("profile_benchmarks.sh") + + +FAKE_JAVA = """#!/usr/bin/env bash +set -euo pipefail + +for argument in "$@"; do + if [[ "${argument}" == "-l" ]]; then + printf 'Benchmarks:\n' + case "${FAKE_LIST_MODE:-one}" in + one) + printf '%s\n' 'org.apache.seatunnel.benchmark.QueueBenchmark.publish' + ;; + two) + printf '%s\n' \ + 'org.apache.seatunnel.benchmark.QueueBenchmark.publish' \ + 'org.apache.seatunnel.benchmark.QueueBenchmark.consume' + ;; + zero) ;; + esac + exit 0 + fi +done + +printf '%s\n' "$@" > "${FAKE_JAVA_ARGS_FILE}" +result_file='' +profiler_directory='' +previous_argument='' +for argument in "$@"; do + if [[ "${previous_argument}" == "-rff" ]]; then + result_file="${argument}" + fi + case "${argument}" in + async:*) + profiler_directory="${argument##*;dir=}" + profiler_directory="${profiler_directory%%;*}" + ;; + esac + previous_argument="${argument}" +done +[[ -n "${result_file}" ]] +printf '[]\n' > "${result_file}" +if [[ -n "${profiler_directory}" ]]; then + trial_directory="${profiler_directory}/fake-benchmark" + mkdir -p "${trial_directory}" + printf '%s\n' \ + '--- Execution profile ---' \ + "Total samples : ${FAKE_ASYNC_SAMPLES:-42}" \ + > "${trial_directory}/summary-cpu.txt" + printf 'jfr\n' > "${trial_directory}/jfr-cpu.jfr" +fi +""" + + +FAKE_JFRCONV = """#!/usr/bin/env bash +set -euo pipefail + +printf '%s\n' "$@" >> "${FAKE_CONVERTER_ARGS_FILE}" +for output_file in "$@"; do :; done +printf '<html>\nf(1)\n</html>\n' > "${output_file}" +""" + + +class ProfileBenchmarksTest(unittest.TestCase): + + def setUp(self): + self.temporary_directory = tempfile.TemporaryDirectory() + self.root = pathlib.Path(self.temporary_directory.name) + self.repository = self.root / "candidate" + target = self.repository / "seatunnel-benchmarks" / "target" + target.mkdir(parents=True) + (target / "benchmarks.jar").touch() + + fake_bin = self.root / "bin" + fake_bin.mkdir() + fake_java = fake_bin / "java" + fake_java.write_text(FAKE_JAVA, encoding="utf-8") + fake_java.chmod(0o755) + + profiler_home = self.root / "async-profiler" + profiler_library = profiler_home / "lib" / "libasyncProfiler.dylib" + profiler_library.parent.mkdir(parents=True) + profiler_library.touch() + profiler_converter = profiler_home / "bin" / "jfrconv" + profiler_converter.parent.mkdir(parents=True) + profiler_converter.write_text(FAKE_JFRCONV, encoding="utf-8") + profiler_converter.chmod(0o755) + + self.java_arguments = self.root / "java-arguments.txt" + self.converter_arguments = self.root / "converter-arguments.txt" + self.environment = os.environ.copy() + self.environment.pop("BENCHMARKS", None) + self.environment.pop("PROFILE_JMH_ARGS", None) + self.environment["FAKE_JAVA_ARGS_FILE"] = str(self.java_arguments) + self.environment["FAKE_CONVERTER_ARGS_FILE"] = str(self.converter_arguments) + self.environment["ASYNC_PROFILER_HOME"] = str(profiler_home) + self.environment["PATH"] = "{}{}{}".format( + fake_bin, os.pathsep, self.environment["PATH"] + ) + + def tearDown(self): + self.temporary_directory.cleanup() + + def run_script(self, *arguments, environment=None): + return subprocess.run( + ["bash", str(SCRIPT), *arguments], + cwd=self.root, + env=environment or self.environment, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + + def test_help_documents_repository_option(self): + result = self.run_script("--help") + + self.assertEqual(0, result.returncode) + self.assertIn("--repository DIR", result.stdout) + + def test_rejects_selector_that_matches_multiple_benchmarks(self): + environment = self.environment.copy() + environment["FAKE_LIST_MODE"] = "two" + + result = self.run_script( + "profile", + "gc", + "--repository", + str(self.repository), + "--benchmark", + ".*", + environment=environment, + ) + + self.assertNotEqual(0, result.returncode) + self.assertIn("matched 2", result.stderr) + self.assertFalse(self.java_arguments.exists()) + + def test_rejects_more_than_one_fork(self): + result = self.run_script( + "profile", + "gc", + "--repository", + str(self.repository), + "--", + "-f", + "2", + ) + + self.assertNotEqual(0, result.returncode) + self.assertIn("require exactly one fork", result.stderr) + self.assertFalse(self.java_arguments.exists()) + + def test_runs_candidate_jar_with_one_fork(self): + output = self.root / "diagnostics" + + result = self.run_script( + "profile", + "gc", + "--repository", + str(self.repository), + "--output", + str(output), + "--", + "-wi", + "0", + "-i", + "1", + "-r", + "1s", + ) + + self.assertEqual(0, result.returncode, result.stderr) + arguments = self.java_arguments.read_text(encoding="utf-8").splitlines() + self.assertIn(str(self.repository / "seatunnel-benchmarks/target/benchmarks.jar"), arguments) + self.assertEqual("1", arguments[arguments.index("-f") + 1]) + self.assertTrue((output / "jmh.log").is_file()) + self.assertTrue((output / "profile-report.json").is_file()) + self.assertTrue((output / "profile-summary.md").is_file()) + + def test_rejects_nonempty_output_directory(self): + output = self.root / "diagnostics" + output.mkdir() + (output / "old-profile.jfr").touch() + + result = self.run_script( + "profile", + "gc", + "--repository", + str(self.repository), + "--output", + str(output), + ) + + self.assertNotEqual(0, result.returncode) + self.assertIn("Output directory is not empty", result.stderr) + self.assertFalse(self.java_arguments.exists()) + + def test_converts_async_jfr_to_forward_and_reverse_flame_graphs(self): + output = self.root / "cpu-diagnostics" + + result = self.run_script( + "profile", + "cpu", + "--repository", + str(self.repository), + "--output", + str(output), + ) + + self.assertEqual(0, result.returncode, result.stderr) + trial = output / "raw" / "fake-benchmark" + self.assertIn("f(1)", (trial / "flame-cpu-forward.html").read_text()) + self.assertIn("f(1)", (trial / "flame-cpu-reverse.html").read_text()) + converter_arguments = self.converter_arguments.read_text(encoding="utf-8") + self.assertIn("--cpu", converter_arguments) + self.assertIn("--reverse", converter_arguments) + + def test_zero_lock_samples_do_not_create_empty_flame_graph(self): + environment = self.environment.copy() + environment["FAKE_ASYNC_SAMPLES"] = "0" + output = self.root / "lock-diagnostics" + + result = self.run_script( + "profile", + "lock", + "--repository", + str(self.repository), + "--output", + str(output), + environment=environment, + ) + + self.assertEqual(0, result.returncode, result.stderr) + self.assertFalse(self.converter_arguments.exists()) + self.assertFalse(list(output.rglob("flame-*.html"))) + summary = (output / "profile-summary.md").read_text(encoding="utf-8") + self.assertIn("Async-profiler samples: `0`", summary) + self.assertIn("no flame graph was generated", summary) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/benchmarks/test_profile_report.py b/tools/benchmarks/test_profile_report.py new file mode 100644 index 0000000000..bc36542ba2 --- /dev/null +++ b/tools/benchmarks/test_profile_report.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. +# + +import json +import pathlib +import tempfile +import unittest + +import profile_report + + +class ProfileReportTest(unittest.TestCase): + + @staticmethod + def write_profile_report(root, directory, **values): + report_directory = root / directory + report_directory.mkdir(parents=True) + report = { + "schema": profile_report.SCHEMA, + "mode": directory, + "command": "profile", + "status": 0, + "gc_metrics": [], + "artifacts": [], + } + report.update(values) + (report_directory / "profile-report.json").write_text( + json.dumps(report), encoding="utf-8" + ) + + def test_ignores_incomplete_jmh_result(self): + with tempfile.TemporaryDirectory() as directory: + result = pathlib.Path(directory) / "result.json" + result.write_text("[", encoding="utf-8") + + self.assertEqual([], profile_report.load_jmh(result)) + + def test_extracts_gc_secondary_metrics(self): + results = [ + { + "benchmark": "org.apache.seatunnel.Queue.publish", + "params": {"capacity": "1024"}, + "secondaryMetrics": { + "gc.alloc.rate.norm": { + "score": 8.0, + "scoreError": 0.1, + "scoreUnit": "B/op", + }, + "gc.count": { + "score": 2.0, + "scoreError": 0.0, + "scoreUnit": "counts", + }, + "unrelated": { + "score": 1.0, + "scoreError": 0.0, + "scoreUnit": "x", + }, + }, + } + ] + + metrics = profile_report.gc_metrics(results) + + self.assertEqual(1, len(metrics)) + self.assertEqual(8.0, metrics[0]["metrics"]["gc.alloc.rate.norm"]["value"]) + self.assertNotIn("unrelated", metrics[0]["metrics"]) + + def test_builds_artifact_list(self): + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + raw = root / "raw" / "benchmark" + raw.mkdir(parents=True) + (raw / "profile.jfr").write_bytes(b"jfr") + artifacts = profile_report.artifacts(root, ()) + + self.assertEqual( + ["raw/benchmark/profile.jfr"], + [artifact["path"] for artifact in artifacts], + ) + + def test_renders_diagnostic_warning(self): + markdown = profile_report.render_markdown( + { + "mode": "cpu", + "command": "profile", + "status": 0, + "schema": profile_report.SCHEMA, + "gc_metrics": [], + "artifacts": [], + } + ) + + self.assertIn("Score comparable with normal benchmarks: `no`", markdown) + self.assertIn("No diagnostic artifacts were produced", markdown) + + def test_reports_async_sample_count_and_nan_marker(self): + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + summary = root / "raw" / "benchmark" / "summary-cpu.txt" + summary.parent.mkdir(parents=True) + summary.write_text( + "--- Execution profile ---\nTotal samples : 42\n", + encoding="utf-8", + ) + + self.assertEqual(42, profile_report.async_sample_count(root)) + + markdown = profile_report.render_markdown( + { + "mode": "cpu", + "command": "profile", + "status": 0, + "schema": profile_report.SCHEMA, + "async_samples": 42, + "gc_metrics": [], + "artifacts": [], + } + ) + self.assertIn("Async-profiler samples: `42`", markdown) + self.assertIn("its `NaN` Score is expected", markdown) + + def test_explains_gc_metrics(self): + markdown = "\n".join( + profile_report.gc_report_lines( + [ + { + "benchmark": "org.apache.seatunnel.Queue.publish", + "params": {}, + "metrics": {}, + } + ] + ) + ) + + self.assertIn("Alloc/op (B/op)", markdown) + self.assertIn("high value alone does not prove a regression", markdown) + self.assertIn("not normalized per operation", markdown) + self.assertIn("not an exact stop-the-world pause-time measurement", markdown) + self.assertIn("Capture JFR separately", markdown) + + def test_renders_compact_workflow_summary_for_cpu_and_jfr(self): + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + self.write_profile_report(root, "profile-cpu", async_samples=9453) + self.write_profile_report( + root, "capture-jfr", mode="jfr", command="capture" + ) + + markdown = profile_report.render_workflow_summary( + diagnostics_dir=root, + profile="cpu", + capture_jfr=True, + target_ref="dev", + pr_number="", + commit="0123456789ab", + benchmark="IntermediateQueueBenchmark.disruptorRecordHandoff$", + java="11", + jmh_args="-f 1 -wi 1 -i 1 -w 1s -r 1s", + artifacts_url="https://example.test/actions/runs/42#artifacts";, + run_id="42", + run_attempt="2", + ) + + self.assertIn("Target: `dev` at `0123456789ab`", markdown) + self.assertIn("Async-profiler samples: `9453`", markdown) + self.assertLess( + markdown.index("### CPU diagnostics"), + markdown.index("### JFR diagnostics"), + ) + self.assertIn("seatunnel-benchmark-profile-cpu-java11-42-2", markdown) + self.assertIn("seatunnel-benchmark-capture-jfr-java11-42-2", markdown) + self.assertIn("https://example.test/actions/runs/42#artifacts";, markdown) + self.assertNotIn("raw/", markdown) + + def test_renders_all_modes_in_order_and_marks_missing_artifact(self): + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + self.write_profile_report(root, "profile-cpu", async_samples=1) + self.write_profile_report(root, "profile-wall", async_samples=2) + self.write_profile_report(root, "profile-lock", async_samples=3) + self.write_profile_report(root, "capture-jfr", mode="jfr") + + markdown = profile_report.render_workflow_summary( + diagnostics_dir=root, + profile="all", + capture_jfr=True, + target_ref="dev", + pr_number="12021", + commit="fedcba987654", + benchmark="Queue.method$", + java="8", + jmh_args="", + artifacts_url="https://example.test/artifacts";, + run_id="7", + run_attempt="1", + ) + + headings = [ + "### CPU diagnostics", + "### Wall diagnostics", + "### Lock diagnostics", + "### GC diagnostics", + "### JFR diagnostics", + ] + positions = [markdown.index(heading) for heading in headings] + self.assertEqual(sorted(positions), positions) + self.assertIn("Target: `PR #12021` at `fedcba987654`", markdown) + self.assertIn("| GC | `not produced` | `not produced` |", markdown) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/benchmarks/test_regression_report.py b/tools/benchmarks/test_regression_report.py new file mode 100644 index 0000000000..3cb3686d4b --- /dev/null +++ b/tools/benchmarks/test_regression_report.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. +# + +import json +import pathlib +import sys +import tempfile +import unittest +from unittest import mock + +import regression_report + + +class RegressionReportTest(unittest.TestCase): + + def test_rejects_unknown_report_schema(self): + with tempfile.TemporaryDirectory() as directory: + path = pathlib.Path(directory) / "report.json" + path.write_text(json.dumps({"schema_version": 2}), encoding="utf-8") + + with self.assertRaisesRegex(ValueError, "Unsupported benchmark report schema"): + regression_report.load_report(path) + + def test_jmh_comparison_uses_medians_and_converts_units(self): + baselines = [ + self.report("dev", self.jmh_metric(1000.0, "ops/s")), + self.report("dev", self.jmh_metric(1.0, "ops/ms")), + ] + candidates = [ + self.report("pr", self.jmh_metric(1.1, "ops/ms")), + self.report("pr", self.jmh_metric(1100.0, "ops/s")), + ] + + markdown = "\n".join( + regression_report.jmh_comparison_lines(baselines, candidates) + ) + + self.assertIn("1.000", markdown) + self.assertIn("1.100", markdown) + self.assertIn("+10.00%", markdown) + self.assertIn("ops/ms", markdown) + + def test_lower_is_better_change_is_reported_as_positive(self): + metric = self.jmh_metric(10.0, "ms/op", direction="lower") + baseline = self.report("dev", metric) + candidate = self.report( + "pr", self.jmh_metric(8.0, "ms/op", direction="lower") + ) + + markdown = "\n".join( + regression_report.jmh_comparison_lines([baseline], [candidate]) + ) + + self.assertIn("+20.00%", markdown) + + def test_report_explains_score_error_and_cv(self): + metric = self.jmh_metric(100.0, "ops/s") + metric.update({"score_error": 5.0, "sample_standard_deviation": 10.0}) + + markdown = "\n".join(regression_report.jmh_report_lines([metric])) + + self.assertIn("confidence-interval half-width", markdown) + self.assertIn("sample standard deviation", markdown) + self.assertIn("5.00%", markdown) + self.assertIn("10.00%", markdown) + + def test_main_writes_markdown_report(self): + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + source = root / "report.json" + output = root / "nested" / "report.md" + source.write_text( + json.dumps(self.report("dev", self.jmh_metric(100.0, "ops/s"))), + encoding="utf-8", + ) + arguments = [ + "regression_report.py", + "--input", + str(source), + "--output", + str(output), + ] + + with mock.patch.object(sys, "argv", arguments): + regression_report.main() + + markdown = output.read_text(encoding="utf-8") + + self.assertIn("## SeaTunnel benchmark report", markdown) + self.assertIn("Queue.publish", markdown) + self.assertTrue(markdown.endswith("\n")) + + @staticmethod + def jmh_metric(value, unit, direction="higher"): + return { + "name": "org.apache.seatunnel.QueueBenchmark.publish[capacity=1024]", + "benchmark": "org.apache.seatunnel.QueueBenchmark.publish", + "kind": "jmh", + "value": value, + "score_error": None, + "sample_standard_deviation": None, + "relative_score_error": None, + "unit": unit, + "direction": direction, + "mode": "thrpt" if direction == "higher" else "avgt", + "params": {"capacity": "1024"}, + "forks": 1, + "samples": [value], + } + + @staticmethod + def report(ref, metric): + return { + "schema_version": 1, + "generated_at": "2026-08-31T00:00:00+00:00", + "source": { + "ref": ref, + "commit": "{}-commit".format(ref), + "run_id": "42", + "suite": "test", + }, + "environment": { + "name": "local", + "java_requested": "8", + "jdk_version": "1.8.0_472", + }, + "metrics": [metric], + "pipeline_correctness": {}, + } + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/benchmarks/test_save_jmh_result.py b/tools/benchmarks/test_save_jmh_result.py new file mode 100644 index 0000000000..fa107a5372 --- /dev/null +++ b/tools/benchmarks/test_save_jmh_result.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. +# + +import json +import pathlib +import sys +import tempfile +import unittest +from unittest import mock + +import save_jmh_result + + +class SaveJmhResultTest(unittest.TestCase): + + def test_normalizes_jmh_samples_and_uncertainty(self): + metrics = save_jmh_result.jmh_metrics( + [ + { + "benchmark": "org.apache.seatunnel.Queue.publish", + "mode": "thrpt", + "forks": 2, + "params": {"workers": "2", "capacity": "1024"}, + "primaryMetric": { + "score": 100.0, + "scoreError": 5.0, + "scoreUnit": "ops/s", + "rawData": [[90.0, 110.0], [100.0]], + }, + } + ] + ) + + self.assertEqual(1, len(metrics)) + metric = metrics[0] + self.assertEqual( + "org.apache.seatunnel.Queue.publish[capacity=1024,workers=2]", + metric["name"], + ) + self.assertEqual([90.0, 110.0, 100.0], metric["samples"]) + self.assertEqual(10.0, metric["sample_standard_deviation"]) + self.assertEqual(0.05, metric["relative_score_error"]) + self.assertEqual("higher", metric["direction"]) + + def test_aggregates_pipeline_medians_correctness_and_clamping(self): + with tempfile.TemporaryDirectory() as directory: + pipeline_dir = pathlib.Path(directory) + samples = [ + self.pipeline_sample("sourceSink-0-0", 1000.0, True, True, 1), + self.pipeline_sample("sourceSink-0-1", 1200.0, False, False, 2), + ] + for index, sample in enumerate(samples): + (pipeline_dir / "{}.json".format(index)).write_text( + json.dumps(sample), encoding="utf-8" + ) + + metrics, correctness = save_jmh_result.pipeline_metrics(pipeline_dir) + + throughput = next( + metric + for metric in metrics + if metric["metric"] == "throughput_rows_per_second" + ) + p50 = next( + metric for metric in metrics if metric["metric"] == "event_time_latency_p50_ms" + ) + p99 = next( + metric for metric in metrics if metric["metric"] == "event_time_latency_p99_ms" + ) + self.assertEqual(1100.0, throughput["value"]) + self.assertFalse(p50["clamped"]) + self.assertTrue(p99["clamped"]) + + values = next(iter(correctness.values())) + self.assertEqual(2, values["sample_count"]) + self.assertEqual(1, values["complete_samples"]) + self.assertEqual(1, values["sustainable_samples"]) + self.assertEqual(2, values["latency_percentiles_clamped_samples"]) + self.assertEqual(3, values["latency_overflow_rows"]) + + def test_main_writes_versioned_report_and_environment(self): + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + jmh = root / "jmh.json" + output = root / "nested" / "report.json" + jmh.write_text( + json.dumps( + [ + { + "benchmark": "org.apache.seatunnel.Queue.publish", + "mode": "thrpt", + "forks": 1, + "params": {}, + "jdkVersion": "1.8.0_472", + "vmName": "OpenJDK 64-Bit Server VM", + "jmhVersion": "1.37", + "primaryMetric": { + "score": 10.0, + "scoreError": 1.0, + "scoreUnit": "ops/s", + "rawData": [[10.0]], + }, + } + ] + ), + encoding="utf-8", + ) + arguments = [ + "save_jmh_result.py", + "--jmh", + str(jmh), + "--output-json", + str(output), + "--ref", + "dev", + "--commit", + "abc123", + "--java", + "8", + "--environment", + "local", + "--run-id", + "42", + "--timestamp", + "2026-08-31T00:00:00+00:00", + ] + + with mock.patch.object(sys, "argv", arguments): + save_jmh_result.main() + + report = json.loads(output.read_text(encoding="utf-8")) + + self.assertEqual(save_jmh_result.SCHEMA_VERSION, report["schema_version"]) + self.assertEqual("2026-08-31T00:00:00+00:00", report["generated_at"]) + self.assertEqual("abc123", report["source"]["commit"]) + self.assertEqual("1.8.0_472", report["environment"]["jdk_version"]) + self.assertEqual(1, len(report["metrics"])) + + @staticmethod + def pipeline_sample(run_id, throughput, complete, sustainable, overflow_rows): + expected_rows = 100 + return { + "run_id": run_id, + "offered_rate_rows_per_second": 1000, + "parallelism": 2, + "payload_size": 256, + "transform_operations": 64, + "processed_rows": expected_rows if complete else expected_rows - 1, + "expected_rows": expected_rows, + "sustainable": sustainable, + "latency_percentiles_clamped": True, + "latency_overflow_rows": overflow_rows, + "throughput_rows_per_second": throughput, + "event_time_latency_p50_ms": 10.0, + "event_time_latency_p95_ms": 100.0, + "event_time_latency_p99_ms": 60001.0, + "event_time_latency_max_ms": 70000.0, + "first_half_p99_ms": 500.0, + "second_half_p99_ms": 60001.0, + "latency_growth_ratio": 2.0, + } + + +if __name__ == "__main__": + unittest.main()
