From 18349a36310c64497a9d34a61376d9a9560d403b Mon Sep 17 00:00:00 2001 From: jiangxt2 Date: Tue, 11 Aug 2026 22:30:28 +0800 Subject: [PATCH] Add an enterprise-candidate MySQL profile Resolve environment-referenced credentials independently before each driver planning request and each worker connection attempt, without serializing resolved values. Add private-CA query planning, an independent query-plan timeout, and fixed redacted MySQL TLS error classification. Authentication, permission, TLS, and missing-credential failures fail closed without transport fallback. Move distributed worker-retry and backend-failure evidence to the minimum-privilege MySQL path with strict TLS, and bind releases to a successful full slow-suite manifest from the exact candidate SHA and workflow run, recording the initial cluster topology in the manifest. Document the logical FE endpoint contract and keep FE election, quorum, and backend failover as deployment concerns. Signed-off-by: jiangxt2 --- .github/workflows/release.yml | 75 +++++++++- .github/workflows/slow-integration.yml | 47 ++++++- CHANGELOG.md | 7 + CONTRIBUTING.md | 5 +- README.md | 59 +++++--- SECURITY.md | 18 ++- doc/source/api/api.md | 6 + doc/source/architecture.md | 8 +- doc/source/compatibility.md | 15 +- doc/source/examples/index.md | 5 +- doc/source/faq.md | 12 ++ doc/source/index.md | 2 +- doc/source/key-concepts.md | 5 + doc/source/quickstart.md | 3 + doc/source/spelling_wordlist.txt | 3 + doc/source/user-guide/configure-transports.md | 10 +- doc/source/user-guide/read-data.md | 5 +- doc/source/user-guide/secure-connections.md | 38 +++++- doc/source/user-guide/troubleshooting.md | 11 +- examples/quickstart.py | 2 +- src/ray_doris/_api.py | 6 + src/ray_doris/_models.py | 66 +++++++-- src/ray_doris/_planner.py | 70 +++++++++- src/ray_doris/_readers.py | 18 ++- src/ray_doris/datasource.py | 6 + tests/integration/conftest.py | 21 ++- tests/integration/test_mysql_read.py | 85 +++++++++++- tests/slow_integration/_cluster.py | 46 ++++++- tests/slow_integration/docker-compose.yml | 5 + tests/slow_integration/docker/generate-tls.sh | 15 +- tests/slow_integration/run.sh | 16 ++- .../test_distributed_cluster.py | 128 ++++++++++++++---- tests/slow_integration/write_result.py | 48 +++++++ tests/unit/test_api.py | 6 + tests/unit/test_datasource.py | 19 +++ tests/unit/test_models.py | 79 +++++++++++ tests/unit/test_planner.py | 117 +++++++++++++++- tests/unit/test_readers.py | 99 ++++++++++++++ tests/unit/test_release.py | 104 ++++++++++++++ tools/check_slow_result.py | 99 ++++++++++++++ 40 files changed, 1293 insertions(+), 96 deletions(-) create mode 100644 tests/slow_integration/write_result.py create mode 100644 tools/check_slow_result.py diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 462cd78..78fabfa 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -6,25 +6,92 @@ on: permissions: contents: read + actions: read jobs: + source: + name: Verify release source + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - run: python3 tools/check_release.py + verify: + needs: source uses: ./.github/workflows/ci.yml build: - needs: verify + needs: [source, verify] runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 + - name: Find exact-SHA slow result + id: slow-run + uses: actions/github-script@v7 with: - fetch-depth: 0 + script: | + const artifactName = `ray-doris-slow-result-${context.sha}`; + const runs = await github.paginate( + github.rest.actions.listWorkflowRunsForRepo, + { + owner: context.repo.owner, + repo: context.repo.repo, + head_sha: context.sha, + status: "success", + per_page: 100, + } + ); + const reusablePath = + `${context.repo.owner}/${context.repo.repo}/.github/workflows/slow-integration.yml@`; + for (const summary of runs) { + const runResponse = await github.rest.actions.getWorkflowRun({ + owner: context.repo.owner, + repo: context.repo.repo, + run_id: summary.id, + }); + const run = runResponse.data; + const directPath = run.path.split("@", 1)[0]; + const direct = directPath === ".github/workflows/slow-integration.yml"; + const reusable = (run.referenced_workflows || []).some((workflow) => + workflow.path.startsWith(reusablePath) && + workflow.sha === context.sha + ); + if (!direct && !reusable) { + continue; + } + const artifacts = await github.rest.actions.listWorkflowRunArtifacts({ + owner: context.repo.owner, + repo: context.repo.repo, + run_id: run.id, + name: artifactName, + per_page: 100, + }); + if (artifacts.data.artifacts.some((artifact) => !artifact.expired)) { + core.setOutput("run-id", String(run.id)); + return; + } + } + core.setFailed(`No successful full slow result exists for ${context.sha}`); + - uses: actions/download-artifact@v4 + with: + name: ray-doris-slow-result-${{ github.sha }} + path: slow-result + github-token: ${{ github.token }} + repository: ${{ github.repository }} + run-id: ${{ steps.slow-run.outputs.run-id }} - uses: astral-sh/setup-uv@v6 with: python-version: "3.12" - run: uv venv --python 3.12 - run: uv pip install build twine - - name: Verify release source and package version - run: .venv/bin/python tools/check_release.py + - name: Verify exact-SHA slow result + run: >- + .venv/bin/python tools/check_slow_result.py + slow-result/slow-result.json + --expected-commit "$GITHUB_SHA" + --expected-run-id "${{ steps.slow-run.outputs.run-id }}" - run: .venv/bin/python -m build - run: .venv/bin/twine check dist/* - uses: actions/upload-artifact@v4 diff --git a/.github/workflows/slow-integration.yml b/.github/workflows/slow-integration.yml index da26f86..b3d5e10 100644 --- a/.github/workflows/slow-integration.yml +++ b/.github/workflows/slow-integration.yml @@ -1,8 +1,36 @@ name: Slow distributed integration on: + workflow_call: + inputs: + profile: + description: Slow integration profile + required: false + default: full + type: string + row_count: + description: Rows loaded into each slow integration table + required: false + default: "10000" + type: string + stress_seconds: + description: Minimum duration of repeated Flight SQL reads + required: false + default: "5" + type: string + be_memory_limit: + description: Memory limit for each Doris BE container + required: false + default: 2g + type: string workflow_dispatch: inputs: + profile: + description: Slow integration profile + required: true + default: full + type: choice + options: [full, core] row_count: description: Rows loaded into each slow integration table required: true @@ -18,6 +46,8 @@ on: required: true default: "2g" type: string + schedule: + - cron: "17 3 * * 0" permissions: contents: read @@ -32,9 +62,10 @@ jobs: runs-on: [self-hosted, linux, x64, ray-doris-slow-it] timeout-minutes: 30 env: - RAY_DORIS_ROW_COUNT: ${{ inputs.row_count }} - RAY_DORIS_STRESS_SECONDS: ${{ inputs.stress_seconds }} - RAY_DORIS_BE_MEMORY_LIMIT: ${{ inputs.be_memory_limit }} + RAY_DORIS_SLOW_PROFILE: ${{ inputs.profile || 'full' }} + RAY_DORIS_ROW_COUNT: ${{ inputs.row_count || '10000' }} + RAY_DORIS_STRESS_SECONDS: ${{ inputs.stress_seconds || '5' }} + RAY_DORIS_BE_MEMORY_LIMIT: ${{ inputs.be_memory_limit || '2g' }} steps: - uses: actions/checkout@v4 - name: Run distributed cluster scenario @@ -47,3 +78,13 @@ jobs: name: ray-doris-slow-it-logs path: ${{ runner.temp }}/ray-doris-slow-it/*.log if-no-files-found: warn + overwrite: true + retention-days: 30 + - uses: actions/upload-artifact@v4 + if: ${{ success() && env.RAY_DORIS_SLOW_PROFILE == 'full' }} + with: + name: ray-doris-slow-result-${{ github.sha }} + path: ${{ runner.temp }}/ray-doris-slow-it/slow-result.json + if-no-files-found: error + overwrite: true + retention-days: 90 diff --git a/CHANGELOG.md b/CHANGELOG.md index 5047e1e..b655df6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,13 @@ All notable changes to this project are documented in this file. - Mark Python 3.9 as Alpha legacy compatibility and Flight SQL and automatic transport selection as experimental. - Validate release ancestry against the `master` default branch. +- Add environment-referenced credentials that resolve independently on the driver and workers + without serializing the resolved value. +- Add a private-CA option and independent timeout for query-plan HTTPS, plus fixed redacted MySQL + TLS setup errors. +- Move distributed worker-retry and backend-failure evidence to the minimum-privilege MySQL path. +- Bind releases to a successful full slow-suite manifest from the exact release commit and workflow + run while keeping FE cluster HA as a deployment responsibility. ## 0.1.0a1 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 7dfcf88..eae2918 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -64,7 +64,10 @@ tests/slow_integration/run.sh Size the dedicated host for the requested container limits; the functional profile keeps each BE at 2 GiB by default. -The slow suite is manual and is not a required check in the default CI workflow. +The slow workflow supports manual, reusable, and scheduled full runs. It is not a required check in +the default CI workflow. A successful full run emits `slow-result.json`; release verification +requires that artifact from a successful workflow run on the exact release SHA. The `core` profile +is diagnostic only and never produces release evidence. ## Documentation checks diff --git a/README.md b/README.md index 2bd049b..4d7baa4 100644 --- a/README.md +++ b/README.md @@ -63,7 +63,7 @@ dataset = read_doris( table="analytics.events", host="doris-fe.example.com", user="ray_reader", - password="...", + password_env="DORIS_PASSWORD", columns=["event_id", "created_at", "score"], filter="score >= 80", tablet_size=32, @@ -99,6 +99,7 @@ read_doris( flight_scheme="grpc", user="root", password="", + password_env=None, columns=None, filter=None, transport="mysql", @@ -106,6 +107,8 @@ read_doris( tablet_size=1, batch_size=10_000, connect_timeout=10.0, + query_plan_timeout=None, + http_ca_file=None, client_kwargs=None, flight_options=None, concurrency=None, @@ -114,21 +117,29 @@ read_doris( ) ``` -`http_scheme` accepts `http` or `https`. HTTPS requires an HTTPS endpoint, commonly a TLS reverse -proxy in front of the Doris FE HTTP API. `flight_scheme` accepts `grpc` or `grpc+tls`; configure -certificates and other ADBC settings with `flight_options`. +`password_env` stores an environment-variable name in the datasource and resolves its value before +each driver request and worker connection attempt. It is mutually exclusive with a non-empty +`password`. The variable name and resolved value are redacted from representations; the resolved +value is never stored in the serialized datasource or ReadTask. -`connect_timeout` is passed to the `_query_plan` HTTP request, each PyMySQL connection attempt, -and the ADBC Flight SQL connect RPC. It does not set a deadline for an established MySQL socket -read or a Flight SQL query/fetch RPC. +`http_scheme` accepts `http` or `https`. HTTPS requires an HTTPS endpoint, commonly a TLS reverse +proxy in front of the Doris FE HTTP API. Set `http_ca_file` for a private CA; hostname verification +remains enabled. `flight_scheme` accepts `grpc` or `grpc+tls`; configure certificates and other ADBC +settings with `flight_options`. + +`connect_timeout` is passed to each PyMySQL connection attempt and the ADBC Flight SQL connect RPC. +`query_plan_timeout` controls the `_query_plan` HTTP request and defaults to `connect_timeout` when +unset. Neither value sets a deadline for an established MySQL socket read or a Flight SQL +query/fetch RPC. Configure those limits explicitly when required: ```python dataset = read_doris( table="analytics.events", host="doris-fe.example.com", - transport="auto", + transport="mysql", connect_timeout=10.0, + query_plan_timeout=30.0, client_kwargs={"read_timeout": 300, "write_timeout": 30}, flight_options={ "adbc.flight.sql.rpc.timeout_seconds.query": "300", @@ -203,12 +214,18 @@ Ray may call `get_read_tasks()` more than once while constructing one read, so e call. Treat an instance as one logical read and create a new instance to discover table or tablet changes made later. This planning cache does not provide snapshot isolation. -Ray serializes datasource configuration to workers. Passwords and transport option values are -redacted from representations and logs, but they still exist in serialized task state. Use this -package only on a trusted Ray cluster and private network, and inject secrets at runtime. Configure -MySQL TLS through `client_kwargs`, set `http_scheme="https"` for a protected query-plan endpoint, -and set `flight_scheme="grpc+tls"` with the required certificate `flight_options` for Flight TLS. -The defaults are unencrypted and must only be used on a trusted private network. +Ray serializes datasource configuration to workers. A literal `password` therefore remains in task +state for compatibility and is suitable only for a trusted cluster. The enterprise-candidate MySQL +profile uses `password_env`, injects the same variable into the driver and every Ray worker, and +resolves it separately for each request or connection attempt without serializing the value. +Transport option values are redacted from representations and logs but remain serialized, so TLS +paths and other sensitive option values still require a trusted Ray control plane and object store. + +Configure MySQL TLS through `client_kwargs`; set `http_scheme="https"` and `http_ca_file` for a +protected query-plan endpoint. The enterprise-candidate profile also uses +`on_query_plan_error="error"`, explicit query-plan/MySQL timeouts, and a minimum-privilege reader. +The defaults are unencrypted and must only be used on a trusted private network. Flight TLS remains +deployment-specific and experimental. The Doris reader account needs access to the FE MySQL and HTTP ports and `SELECT` on the target internal-catalog table. Flight reads additionally need the FE Flight SQL port. The `_query_plan` @@ -218,6 +235,11 @@ Tablet planning and task execution do not provide snapshot isolation. Concurrent therefore produce a result that reflects different moments across splits. If a Ray task fails after reading part of a split, Ray can retry the whole task; the connector does not resume a partial split. +Configure one logical FE hostname that is valid for both HTTPS and MySQL TLS. `ray-doris` validates +and uses that endpoint but doesn't discover FE members or implement leader election, quorum, health +checks, or cross-endpoint failover. Production deployments must provide and validate those HA +properties in Doris and their external load balancer. + The required Doris 4.0.6 integration suite uses the default HTTP endpoint. The distributed suite uses the same fixed Doris version, validates native MySQL TLS, and validates certificate-checked HTTPS through an HAProxy ingress that forwards to the FE HTTP endpoint. It does not enable Doris @@ -284,8 +306,9 @@ The opt-in slow suite runs the following isolated topology: - certificate-verified HTTPS query planning at the ingress and native Doris MySQL TLS; - explicit Arrow Flight SQL reads, with no automatic MySQL fallback; - per-BE Flight session and byte counters proving that all three BE services receive traffic; -- a Ray worker failure after the first Flight block and a retry on another worker; -- a Doris BE failure, proxy health removal, and a complete read from surviving replicas; +- a minimum-privilege MySQL read distributed across all three Ray workers; +- a Ray worker failure after the first MySQL block and a complete-split retry on another worker; +- a Doris BE failure and a complete MySQL read from surviving replicas; - 10,000 rows by default and repeated checksum-validated Flight reads for at least five seconds. It is excluded from the default pytest discovery paths and from the regular CI workflow. Run it @@ -326,6 +349,10 @@ Size the dedicated host for the requested container limits. The script refuses t existing `ray-doris-it` Compose project, preserves pytest, Ray, Doris, and HAProxy logs, and removes only the resources created by that exact project. +Successful `full` runs write `slow-result.json`. The scheduled/reusable workflow uploads it under an +artifact name bound to the tested commit; release verification accepts only a successful full +manifest whose commit and workflow run ID exactly match the downloaded artifact source. + See [CONTRIBUTING.md](CONTRIBUTING.md) for the complete checks. Build and validate the documentation with the commands in the diff --git a/SECURITY.md b/SECURITY.md index 20994ea..62fc3aa 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -8,11 +8,17 @@ impact, and any suggested mitigation. ## Credential handling -`ray-doris` sends credentials to Ray workers as part of datasource configuration. Run it only on a -trusted Ray cluster and network. Inject secrets at runtime, grant the Doris account only `SELECT` -on required tables, and avoid placing credentials in source code, logs, or issue reports. +A literal `password` remains in Ray's serialized datasource and task state for compatibility. The +enterprise-candidate MySQL profile instead uses `password_env`: only the environment-variable name +is serialized, and the driver and each worker resolve its value immediately before connecting. +Inject the same variable into every eligible Ray process, run only on a trusted Ray control plane +and object store, grant the Doris account only `SELECT` on required tables, and avoid placing +credentials in source code, logs, or issue reports. The default MySQL, query-plan HTTP, and Flight URI schemes are unencrypted. Configure MySQL TLS -through `client_kwargs`, use `http_scheme="https"` with a trusted TLS endpoint for query planning, -and use `flight_scheme="grpc+tls"` with certificate settings in `flight_options`. Do not send -production credentials over the default schemes outside a trusted private network. +through `client_kwargs`, use `http_scheme="https"` with `http_ca_file` when a private CA protects +query planning, and use `flight_scheme="grpc+tls"` with certificate settings in `flight_options`. +Do not send production credentials over the default schemes outside a trusted private network. + +The connector accepts one logical FE hostname. Doris and the deployment platform remain +responsible for FE membership, leader election, quorum, health checks, and load-balancer failover. diff --git a/doc/source/api/api.md b/doc/source/api/api.md index bf4b9bc..316e798 100644 --- a/doc/source/api/api.md +++ b/doc/source/api/api.md @@ -12,6 +12,12 @@ This page documents the curated public package surface. Signatures and docstring The connector uses Ray's documented Datasource extension interfaces without importing `ray.data._internal`. Ray classifies `ReadTask` as DeveloperAPI, so compatibility is limited to the tested Ray minor window. MySQL is the production-candidate transport; Flight SQL and `auto` are experimental. +For the enterprise-candidate MySQL profile, use `password_env` rather than a literal password, +strict MySQL TLS in `client_kwargs`, HTTPS with `http_ca_file`, an explicit +`query_plan_timeout`, and `on_query_plan_error="error"`. The environment credential is resolved +again in every driver request and worker attempt; its value isn't stored in the datasource or +ReadTask payload. + ## Read a Doris table (ray-doris-api-read-doris)= diff --git a/doc/source/architecture.md b/doc/source/architecture.md index 6c4c13e..2c4bb4a 100644 --- a/doc/source/architecture.md +++ b/doc/source/architecture.md @@ -38,7 +38,7 @@ Only `read_doris`, `DorisDatasource`, and the exception hierarchy are public pac ## Plan on the driver -The driver performs two metadata operations. MySQL `DESCRIBE` produces column names, types, and nullability. An authenticated HTTP POST to `/_query_plan` produces predicate-pruned tablet IDs. +The driver performs two metadata operations. MySQL `DESCRIBE` produces column names, types, and nullability. An authenticated HTTP POST to `/_query_plan` produces predicate-pruned tablet IDs. When `password_env` is configured, each operation resolves the credential immediately before its own network request. A custom `http_ca_file` is loaded into a hostname-verifying TLS context for query planning. The query-plan client disables redirects and validates both HTTP failures and Doris's body envelope. Doris can report application errors inside an HTTP 200 response, so HTTP status alone isn't enough. @@ -50,7 +50,7 @@ The MySQL reader distinguishes normal EOF from consumer abort. Normal completion MySQL is the production-candidate data path. Flight SQL and worker-local `auto` selection remain experimental and aren't part of a stable compatibility profile. -Transport option mappings are deep-copied into immutable tuple storage so caller mutation can't change a constructed datasource. Construction also verifies the actual configuration with Ray's worker serialization protocol and rejects unsupported values before network access. The configuration representation exposes option keys for diagnosis but redacts the password, filter, and every option value. +Transport option mappings are deep-copied into immutable tuple storage so caller mutation can't change a constructed datasource. Construction also verifies the actual configuration with Ray's worker serialization protocol and rejects unsupported values before network access. The configuration representation exposes option keys for diagnosis but redacts the password, password environment name, HTTP CA path, filter, and every option value. An environment credential is resolved per process and never written back into the serialized configuration. ## Keep SQL generation narrow @@ -71,6 +71,10 @@ Configuration, schema, planning, authentication, permission, and worker read fai Automatic transport fallback uses an internal setup-only exception. The Flight reader translates eligible dependency, I/O, timeout, and unsupported-operation failures before a reader exists. It doesn't translate stream and conversion errors into fallback signals. +The connector accepts one logical FE host and never discovers cluster members. Doris and the +deployment platform own FE election, quorum, health checking, and load-balancer failover. The +connector owns TLS verification and failure classification for the configured endpoint. + ## Define the consistency boundary The connector doesn't coordinate a Doris transaction across tablet tasks. Metadata discovery, query planning, and split reads are separate requests. Task retry is at-least-once at the split level because a replacement task re-executes the complete split. diff --git a/doc/source/compatibility.md b/doc/source/compatibility.md index 062c2cf..54c15b3 100644 --- a/doc/source/compatibility.md +++ b/doc/source/compatibility.md @@ -20,7 +20,10 @@ Continuous integration covers these combinations: | 3.10 | 2.55.1 | Unit and Ray signature compatibility tests | | 3.12 | 2.56.1 | Unit tests and required Doris 4.0.6 integration tests | -The optional distributed suite uses Python 3.12, Ray 2.55.1, and Doris 4.0.6. It runs one Ray head with no scheduling CPUs, three one-CPU Ray workers, one Doris frontend, three Doris backends, and a TLS and Flight ingress. +The optional distributed suite uses Python 3.12, Ray 2.55.1, and Doris 4.0.6. It runs one Ray head +with no scheduling CPUs, three one-CPU Ray workers, one Doris frontend, three Doris backends, and a +TLS and Flight ingress. The enterprise-candidate evidence uses the minimum-privilege MySQL reader; +Flight remains an experimental regression path. ## Understand the Doris target @@ -33,8 +36,11 @@ Doris 4.0.6 is the fixed real-infrastructure target for required and distributed - Multiple Ray worker and Doris backend participation. - Ray worker retry and Doris backend failure with replicated tablets. - Native MySQL TLS and certificate-validated HTTPS through the test ingress. +- Environment-referenced credentials resolved independently on the driver and workers. -The suite doesn't verify every Doris release, storage model, deployment proxy, authentication provider, or Flight TLS endpoint. +The suite doesn't verify every Doris release, storage model, deployment proxy, authentication +provider, or Flight TLS endpoint. It uses one logical FE hostname but doesn't certify Doris FE +leader election, quorum, multi-FE failover, or an external load balancer's backend policy. ## Review Python and Flight limits @@ -61,4 +67,9 @@ MySQL is the production-candidate transport. Flight SQL and `auto` remain experi The required integration suite uses the default frontend HTTP query-plan endpoint. The distributed suite verifies certificate-validated HTTPS through HAProxy because Doris 4.0.6 native frontend HTTPS has a Jetty WebSocket startup regression in this test topology. +The enterprise-candidate MySQL profile requires `password_env`, strict MySQL TLS, HTTPS query +planning with hostname verification, `on_query_plan_error="error"`, and explicit query-plan and +MySQL socket timeouts. A release candidate also requires a successful full slow manifest bound to +the exact release SHA and workflow run. + Doris 4.0.6 advertises plaintext Flight `grpc` endpoints in the distributed suite. Flight stays on an isolated Compose network. The test doesn't establish a positive `grpc+tls` Doris server compatibility claim. diff --git a/doc/source/examples/index.md b/doc/source/examples/index.md index c137cd0..c8d3f29 100644 --- a/doc/source/examples/index.md +++ b/doc/source/examples/index.md @@ -58,8 +58,11 @@ dataset = read_doris( mysql_port=9030, http_port=8443, http_scheme="https", + http_ca_file="/etc/doris-tls/ca.pem", user="ray_reader", - password="...", + password_env="DORIS_PASSWORD", + query_plan_timeout=30, + on_query_plan_error="error", client_kwargs={ "ssl": { "ca": "/etc/doris-tls/ca.pem", diff --git a/doc/source/faq.md b/doc/source/faq.md index 1743bd6..b50c792 100644 --- a/doc/source/faq.md +++ b/doc/source/faq.md @@ -58,6 +58,18 @@ No. Ray can retry the complete read task according to its remote arguments. `ray The account needs `SELECT_PRIV` on the target internal-catalog table and network access to the frontend MySQL and HTTP ports. Add Flight endpoint access for Flight reads. The connector doesn't require administrator-only tablet metadata commands. +## Does ray-doris serialize my password? + +A literal `password` remains in serialized task state for compatibility. With `password_env`, only +the variable name is serialized; the driver and each worker resolve its value immediately before a +network connection. Inject the same variable into every eligible Ray process. + +## Does ray-doris manage Doris FE failover? + +No. Configure one logical FE hostname. The connector verifies TLS and reopens connections for new +planning and task attempts, but Doris and your external load balancer own FE discovery, leader +election, quorum, health checks, and backend failover. + ## Where do I install the Flight extra? Install the extra on the driver and in every environment that can execute a Ray read task. The driver validates explicit `flight` requests, and worker-side dependency availability determines whether `auto` can use Flight. diff --git a/doc/source/index.md b/doc/source/index.md index ba3e276..1315125 100644 --- a/doc/source/index.md +++ b/doc/source/index.md @@ -25,7 +25,7 @@ dataset = read_doris( table="analytics.events", host="doris-fe.example.com", user="ray_reader", - password="...", + password_env="DORIS_PASSWORD", columns=["event_id", "created_at", "score"], filter="score >= 80", tablet_size=32, diff --git a/doc/source/key-concepts.md b/doc/source/key-concepts.md index c5d374a..9a6b64a 100644 --- a/doc/source/key-concepts.md +++ b/doc/source/key-concepts.md @@ -54,6 +54,7 @@ These settings control different layers: | `concurrency` | Ray execution | Limits the number of read tasks that execute at the same time | | `batch_size` | Worker output | Bounds MySQL `fetchmany()` batches and slices Flight results into Ray blocks | | `override_num_blocks` | Ray output planning | Requests an output block count independently of the source tablet count | +| `query_plan_timeout` | Driver planning | Bounds query-plan HTTP request I/O; defaults to `connect_timeout` | See [Tune parallelism](user-guide/tune-parallelism.md) before changing more than one setting. @@ -78,3 +79,7 @@ An empty `partitions` object from a successful query plan represents a valid emp Tablet planning and split execution don't provide snapshot isolation. Concurrent writes can make different tasks observe different moments. If Ray retries a failed task after it emitted part of a split, the replacement task reads the complete split again. `ray-doris` doesn't resume from a partial row offset. Create a new {ref}`DorisDatasource ` for each logical read. A datasource instance caches its first schema and tablet discovery result because Ray can request read tasks more than once while constructing a Dataset. The cache avoids repeated planning calls but doesn't create a database snapshot. + +`password_env` preserves the driver/worker boundary: the serialized task contains only the +environment-variable name, while each process resolves the current value before opening its own +connection. Ray retries therefore resolve the credential again before re-executing a complete split. diff --git a/doc/source/quickstart.md b/doc/source/quickstart.md index 8b5ccb1..750917c 100644 --- a/doc/source/quickstart.md +++ b/doc/source/quickstart.md @@ -64,6 +64,9 @@ export DORIS_PASSWORD= python examples/quickstart.py ``` +The example passes `password_env="DORIS_PASSWORD"`; it doesn't read the value into datasource +configuration. In a cluster, inject that variable into the driver and all Ray workers. + The example uses the public {ref}`read_doris ` entry point: ```{literalinclude} ../../examples/quickstart.py diff --git a/doc/source/spelling_wordlist.txt b/doc/source/spelling_wordlist.txt index f48a041..1302e4b 100644 --- a/doc/source/spelling_wordlist.txt +++ b/doc/source/spelling_wordlist.txt @@ -2,6 +2,7 @@ args autodoc backend backends +balancer charset Datasink datasource @@ -16,11 +17,13 @@ DorisPlanningError DorisReadConfig DorisReadError DorisSchemaError +env failover frontend grpc http https +hostname kwargs mysql nullability diff --git a/doc/source/user-guide/configure-transports.md b/doc/source/user-guide/configure-transports.md index 110d56e..61bf331 100644 --- a/doc/source/user-guide/configure-transports.md +++ b/doc/source/user-guide/configure-transports.md @@ -41,6 +41,7 @@ dataset = read_doris( table="analytics.events", host="doris-fe.example.com", transport="mysql", + password_env="DORIS_PASSWORD", batch_size=10_000, client_kwargs={"read_timeout": 300, "write_timeout": 30}, ) @@ -80,20 +81,23 @@ Setting `flight_scheme="grpc+tls"` makes `auto` fail closed. A TLS setup failure ## Configure timeout scopes -`connect_timeout` applies to three setup operations: +`connect_timeout` applies to two setup operations: -- The frontend `_query_plan` HTTP request. - Each PyMySQL connection attempt. - The ADBC Flight SQL connect remote procedure call (RPC). +`query_plan_timeout` applies to the frontend `_query_plan` HTTP request. When it is `None`, the +request keeps the backward-compatible `connect_timeout` value. + It doesn't set a deadline for an established MySQL socket read or a Flight query and fetch. Configure those execution deadlines with `client_kwargs` and `flight_options`: ```python dataset = read_doris( table="analytics.events", host="doris-fe.example.com", - transport="auto", + transport="mysql", connect_timeout=10.0, + query_plan_timeout=30.0, client_kwargs={"read_timeout": 300, "write_timeout": 30}, flight_options={ "adbc.flight.sql.rpc.timeout_seconds.query": "300", diff --git a/doc/source/user-guide/read-data.md b/doc/source/user-guide/read-data.md index aff07d7..7fa1c5b 100644 --- a/doc/source/user-guide/read-data.md +++ b/doc/source/user-guide/read-data.md @@ -21,10 +21,13 @@ dataset = read_doris( table="analytics.events", host="doris-fe.example.com", user="ray_reader", - password="...", + password_env="DORIS_PASSWORD", ) ``` +`password_env` is the recommended credential boundary for distributed use. See +[Secure connections](secure-connections.md) for driver/worker injection and TLS requirements. + Database, table, and column identifiers must start with a letter or underscore and then contain only letters, numbers, underscores, or dollar signs. `ray-doris` quotes validated identifiers in generated SQL. Three-part names such as `internal.analytics.events` fail before network access. External catalogs, joins, subqueries in the table argument, and arbitrary `SELECT` statements aren't supported. diff --git a/doc/source/user-guide/secure-connections.md b/doc/source/user-guide/secure-connections.md index 997e092..6c431ab 100644 --- a/doc/source/user-guide/secure-connections.md +++ b/doc/source/user-guide/secure-connections.md @@ -14,7 +14,26 @@ TLS configuration doesn't change transport maturity: MySQL is the production-can ## Understand the credential boundary -Ray serializes datasource configuration into worker task state. `repr()` and connector logs redact passwords and option values, but the real values remain in serialized state so workers can connect to Doris. +Ray serializes datasource configuration into worker task state. A literal `password` is redacted +from `repr()` and connector logs but remains in that serialized state for compatibility. + +For the enterprise-candidate MySQL profile, pass the name of an environment variable instead: + +```python +dataset = read_doris( + table="analytics.events", + host="doris.example.com", + user="ray_reader", + password_env="DORIS_PASSWORD", +) +``` + +Only the environment-variable name is serialized. The driver resolves its value before each +`DESCRIBE` or query-plan request, and a worker resolves it again before every split connection +attempt. Inject the same variable into the driver and every Ray worker, including replacement +workers. A missing variable fails before that process opens a network connection; an empty value +preserves Doris's empty-password behavior. Don't set both a non-empty literal `password` and +`password_env`. Run `ray-doris` only on a trusted Ray cluster. Inject credentials at runtime, restrict access to Ray logs and object storage, and don't put secrets in source code, documentation, issue reports, or persistent Dataset references. @@ -35,7 +54,7 @@ dataset = read_doris( table="analytics.events", host="doris.example.com", user="ray_reader", - password="...", + password_env="DORIS_PASSWORD", client_kwargs={ "ssl": { "ca": "/etc/doris-tls/ca.pem", @@ -59,15 +78,28 @@ dataset = read_doris( host="doris.example.com", http_port=8443, http_scheme="https", + http_ca_file="/etc/doris-tls/ca.pem", user="ray_reader", - password="...", + password_env="DORIS_PASSWORD", + query_plan_timeout=30, ) ``` The client refuses authenticated POST redirects. Configure the final endpoint directly so a redirect can't change the request method or expose the `Authorization` header. +`http_ca_file` loads a private CA with Python's default TLS context. Certificate and hostname +verification remain enabled; there is no trust-all or hostname-bypass option. Leave it unset to use +the process default trust store. It is valid only with `http_scheme="https"`. + A TLS validation failure raises {ref}`DorisConfigurationError ` and never becomes a single-task planning fallback. +## Define the FE availability boundary + +Use one logical hostname for the query-plan and MySQL endpoints. `ray-doris` validates that +hostname and reopens connections for each planning or split attempt, but it doesn't discover FE +members or manage leader election, quorum, health checks, or endpoint failover. Provide those +properties with Doris and an external load balancer, and test that deployment independently. + ## Configure Flight TLS Set `flight_scheme="grpc+tls"` and pass the Arrow Database Connectivity (ADBC) certificate options required by your endpoint: diff --git a/doc/source/user-guide/troubleshooting.md b/doc/source/user-guide/troubleshooting.md index 718873d..7cf31a8 100644 --- a/doc/source/user-guide/troubleshooting.md +++ b/doc/source/user-guide/troubleshooting.md @@ -20,8 +20,13 @@ Start with the public exception type and the stage named in its message. Connect - `tablet_size`, `batch_size`, and Ray parallelism values are positive integers. - Timeout values are finite and positive. - Option mappings don't override connector-managed connection fields. +- `password_env` is a portable environment name and is available in both driver and worker processes. +- `http_ca_file` is non-empty and used only with `http_scheme="https"`. -An HTTPS certificate failure and an authenticated query-plan redirect also use this exception because you must change the configuration before planning can continue safely. +An HTTPS certificate failure, a MySQL TLS setup failure during driver planning, and an authenticated +query-plan redirect also use this exception because you must change the configuration before +planning can continue safely. Messages don't include environment-variable names, CA paths, or +driver exception text. ## Fix authentication and permission errors @@ -51,7 +56,9 @@ A Flight stream, schema, authentication, permission, or conversion error doesn't ## Diagnose slow or stalled reads -`connect_timeout` doesn't stop an established query. Configure PyMySQL `read_timeout` and `write_timeout` or Arrow Database Connectivity (ADBC) Flight query and fetch timeouts. Then compare Ray task duration with Doris query and backend metrics. +`query_plan_timeout` bounds query-plan request I/O. `connect_timeout` doesn't stop an established +query. Configure PyMySQL `read_timeout` and `write_timeout` or Arrow Database Connectivity (ADBC) +Flight query and fetch timeouts. Then compare Ray task duration with Doris query and backend metrics. Reduce `concurrency` when Doris carries too much load. Increase `tablet_size` when Ray schedules too many short tasks. Don't interpret `override_num_blocks` as a Doris connection limit. diff --git a/examples/quickstart.py b/examples/quickstart.py index fef8ec8..39fce79 100644 --- a/examples/quickstart.py +++ b/examples/quickstart.py @@ -11,7 +11,7 @@ http_port=int(os.environ.get("DORIS_HTTP_PORT", "8030")), http_scheme=os.environ.get("DORIS_HTTP_SCHEME", "http"), user=os.environ.get("DORIS_USER", "root"), - password=os.environ.get("DORIS_PASSWORD", ""), + password_env="DORIS_PASSWORD", columns=["event_id", "created_at", "score"], filter="score >= 80", tablet_size=32, diff --git a/src/ray_doris/_api.py b/src/ray_doris/_api.py index f6f636a..9896bf8 100644 --- a/src/ray_doris/_api.py +++ b/src/ray_doris/_api.py @@ -22,6 +22,7 @@ def read_doris( flight_scheme: FlightScheme = "grpc", user: str = "root", password: str = "", + password_env: Optional[str] = None, columns: Optional[Sequence[str]] = None, filter: Optional[str] = None, transport: Transport = "mysql", @@ -29,6 +30,8 @@ def read_doris( tablet_size: int = 1, batch_size: int = 10_000, connect_timeout: float = 10.0, + query_plan_timeout: Optional[float] = None, + http_ca_file: Optional[str] = None, client_kwargs: Optional[Mapping[str, Any]] = None, flight_options: Optional[Mapping[str, Any]] = None, concurrency: Optional[int] = None, @@ -46,6 +49,7 @@ def read_doris( flight_scheme=flight_scheme, user=user, password=password, + password_env=password_env, columns=columns, filter=filter, transport=transport, @@ -53,6 +57,8 @@ def read_doris( tablet_size=tablet_size, batch_size=batch_size, connect_timeout=connect_timeout, + query_plan_timeout=query_plan_timeout, + http_ca_file=http_ca_file, client_kwargs=client_kwargs, flight_options=flight_options, ) diff --git a/src/ray_doris/_models.py b/src/ray_doris/_models.py index 98b1baa..8e29e61 100644 --- a/src/ray_doris/_models.py +++ b/src/ray_doris/_models.py @@ -3,6 +3,8 @@ from __future__ import annotations import math +import os +import re from collections.abc import Mapping as MappingABC from copy import deepcopy from dataclasses import dataclass, field @@ -19,6 +21,7 @@ _ADBC_CONNECT_TIMEOUT_OPTION = "adbc.flight.sql.rpc.timeout_seconds.connect" _MAX_CONNECT_TIMEOUT_SECONDS = 31_536_000 +_ENVIRONMENT_VARIABLE_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") _RESERVED_MYSQL_OPTIONS = { "charset", @@ -79,10 +82,12 @@ class DorisReadConfig: table: QualifiedTable user: str password: str = field(default="", repr=False) + password_env: Optional[str] = field(default=None, repr=False) mysql_port: int = 9030 http_port: int = 8030 flight_port: int = 8070 http_scheme: HttpScheme = "http" + http_ca_file: Optional[str] = field(default=None, repr=False) flight_scheme: FlightScheme = "grpc" columns: Optional[Tuple[str, ...]] = None filter: Optional[str] = None @@ -91,6 +96,7 @@ class DorisReadConfig: batch_size: int = 10_000 on_query_plan_error: QueryPlanPolicy = "single_task" connect_timeout: float = 10.0 + query_plan_timeout: Optional[float] = None client_options: Tuple[Tuple[str, Any], ...] = field(default_factory=tuple, repr=False) flight_options: Tuple[Tuple[str, Any], ...] = field(default_factory=tuple, repr=False) @@ -102,6 +108,13 @@ def __post_init__(self) -> None: raise DorisConfigurationError("user must not be empty") if not isinstance(self.password, str): raise DorisConfigurationError("password must be a string") + if self.password_env is not None and ( + not isinstance(self.password_env, str) + or _ENVIRONMENT_VARIABLE_PATTERN.fullmatch(self.password_env) is None + ): + raise DorisConfigurationError("password_env must be a portable environment name") + if self.password and self.password_env is not None: + raise DorisConfigurationError("password and password_env are mutually exclusive") for name, port in ( ("mysql_port", self.mysql_port), ("http_port", self.http_port), @@ -121,20 +134,18 @@ def __post_init__(self) -> None: or self.batch_size <= 0 ): raise DorisConfigurationError("batch_size must be a positive integer") - if ( - isinstance(self.connect_timeout, bool) - or not isinstance(self.connect_timeout, (int, float)) - or self.connect_timeout <= 0 - or self.connect_timeout > _MAX_CONNECT_TIMEOUT_SECONDS - or not math.isfinite(self.connect_timeout) - ): - raise DorisConfigurationError( - "connect_timeout must be finite, positive, and at most 31536000 seconds" - ) + self._validate_timeout("connect_timeout", self.connect_timeout) + if self.query_plan_timeout is not None: + self._validate_timeout("query_plan_timeout", self.query_plan_timeout) if self.transport not in ("auto", "mysql", "flight"): raise DorisConfigurationError(f"unsupported transport: {self.transport!r}") if self.http_scheme not in ("http", "https"): raise DorisConfigurationError(f"unsupported HTTP scheme: {self.http_scheme!r}") + if self.http_ca_file is not None: + if not isinstance(self.http_ca_file, str) or not self.http_ca_file: + raise DorisConfigurationError("http_ca_file must be a non-empty string") + if self.http_scheme != "https": + raise DorisConfigurationError("http_ca_file requires http_scheme='https'") if self.flight_scheme not in ("grpc", "grpc+tls"): raise DorisConfigurationError(f"unsupported Flight SQL scheme: {self.flight_scheme!r}") if self.on_query_plan_error not in ("single_task", "error"): @@ -153,6 +164,19 @@ def __post_init__(self) -> None: if any(not isinstance(value, str) for _, value in self.flight_options): raise DorisConfigurationError("flight_options values must be strings") + @staticmethod + def _validate_timeout(name: str, value: object) -> None: + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or value <= 0 + or value > _MAX_CONNECT_TIMEOUT_SECONDS + or not math.isfinite(value) + ): + raise DorisConfigurationError( + f"{name} must be finite, positive, and at most 31536000 seconds" + ) + def _validate_mysql_timeouts(self) -> None: options = dict(self.client_options) for name in ("read_timeout", "write_timeout"): @@ -225,6 +249,22 @@ def mysql_options(self) -> Mapping[str, Any]: """Return a fresh mapping for a PyMySQL connection.""" return deepcopy(dict(self.client_options)) + def resolve_password(self) -> str: + """Resolve a per-process credential without storing it in the config.""" + if self.password_env is None: + return self.password + try: + return os.environ[self.password_env] + except KeyError: + raise DorisConfigurationError( + "configured password environment variable is unavailable" + ) from None + + @property + def effective_query_plan_timeout(self) -> float: + """Return the configured query-plan I/O timeout.""" + return self.connect_timeout if self.query_plan_timeout is None else self.query_plan_timeout + def adbc_options(self) -> Mapping[str, Any]: """Return a fresh mapping for an ADBC Flight SQL connection.""" return deepcopy(dict(self.flight_options)) @@ -232,17 +272,21 @@ def adbc_options(self) -> Mapping[str, Any]: def __repr__(self) -> str: """Return a representation that never exposes credentials or option values.""" rendered_filter = "None" if self.filter is None else "" + rendered_password_env = "None" if self.password_env is None else "" + rendered_http_ca = "None" if self.http_ca_file is None else "" return ( "DorisReadConfig(" f"host={self.host!r}, table={self.table!r}, user={self.user!r}, " - "password=, " + f"password=, password_env={rendered_password_env}, " f"mysql_port={self.mysql_port}, http_port={self.http_port}, " f"flight_port={self.flight_port}, http_scheme={self.http_scheme!r}, " + f"http_ca_file={rendered_http_ca}, " f"flight_scheme={self.flight_scheme!r}, columns={self.columns!r}, " f"filter={rendered_filter}, transport={self.transport!r}, " f"tablet_size={self.tablet_size}, batch_size={self.batch_size}, " f"on_query_plan_error={self.on_query_plan_error!r}, " f"connect_timeout={self.connect_timeout}, " + f"query_plan_timeout={self.query_plan_timeout}, " f"client_options={tuple(key for key, _ in self.client_options)!r}, " f"flight_options={tuple(key for key, _ in self.flight_options)!r})" ) diff --git a/src/ray_doris/_planner.py b/src/ray_doris/_planner.py index 508fca1..485bf1b 100644 --- a/src/ray_doris/_planner.py +++ b/src/ray_doris/_planner.py @@ -73,8 +73,26 @@ def redirect_request( return None -def _open_query_plan_request(request: urllib.request.Request, timeout: float) -> Any: - return urllib.request.build_opener(_NoRedirectHandler()).open(request, timeout=timeout) +def _open_query_plan_request( + request: urllib.request.Request, + timeout: float, + ssl_context: Optional[ssl.SSLContext], +) -> Any: + handlers: List[Any] = [_NoRedirectHandler()] + if ssl_context is not None: + handlers.append(urllib.request.HTTPSHandler(context=ssl_context)) + return urllib.request.build_opener(*handlers).open(request, timeout=timeout) + + +def _query_plan_ssl_context(config: DorisReadConfig) -> Optional[ssl.SSLContext]: + if config.http_scheme != "https": + return None + try: + return ssl.create_default_context(cafile=config.http_ca_file) + except (OSError, ValueError): + raise DorisConfigurationError( + f"Doris query-plan TLS configuration is invalid for {_table_context(config)}" + ) from None def _contains_tls_error(exc: BaseException) -> bool: @@ -97,13 +115,36 @@ def _contains_tls_error(exc: BaseException) -> bool: return False +def _contains_mysql_tls_error(exc: pymysql.MySQLError) -> bool: + if _contains_tls_error(exc): + return True + markers = ( + "certificate verify failed", + "certificate_verify_failed", + "hostname mismatch", + "ssl handshake", + "tls handshake", + ) + return any( + marker in argument.lower() + for argument in exc.args + if isinstance(argument, str) + for marker in markers + ) + + +def _mysql_tls_is_configured(config: DorisReadConfig) -> bool: + options = config.mysql_options() + return "ssl" in options or any(key.startswith("ssl_") for key in options) + + def _mysql_connection_kwargs(config: DorisReadConfig, *, streaming: bool) -> Dict[str, Any]: options = dict(config.mysql_options()) kwargs: Dict[str, Any] = { "host": config.host, "port": config.mysql_port, "user": config.user, - "password": config.password, + "password": config.resolve_password(), "database": config.table.database, "charset": "utf8mb4", "connect_timeout": config.connect_timeout, @@ -127,18 +168,24 @@ def fetch_tablet_ids(self, sql: str) -> Tuple[int, ...]: f"{config.http_scheme}://{config.host}:{config.http_port}/api/" f"{config.table.database}/{config.table.table}/_query_plan" ) + password = config.resolve_password() + ssl_context = _query_plan_ssl_context(config) request = urllib.request.Request( url, data=json.dumps({"sql": sql}).encode("utf-8"), headers={ "Authorization": "Basic " - + base64.b64encode(f"{config.user}:{config.password}".encode()).decode(), + + base64.b64encode(f"{config.user}:{password}".encode()).decode(), "Content-Type": "application/json", }, method="POST", ) try: - with _open_query_plan_request(request, timeout=config.connect_timeout) as response: + with _open_query_plan_request( + request, + timeout=config.effective_query_plan_timeout, + ssl_context=ssl_context, + ) as response: payload_bytes = response.read() except urllib.error.HTTPError as exc: if 300 <= exc.code < 400: @@ -323,9 +370,22 @@ def _describe_schema(self) -> pa.Schema: cursor.close() finally: connection.close() + except DorisConfigurationError: + raise + except OSError: + context = _table_context(self._config) + if _mysql_tls_is_configured(self._config): + raise DorisConfigurationError( + f"Doris MySQL TLS configuration is invalid for {context}" + ) from None + raise DorisPlanningError(f"failed to discover Doris schema for {context}") from None except pymysql.MySQLError as exc: code = _mysql_error_code(exc) context = _table_context(self._config) + if _contains_mysql_tls_error(exc): + raise DorisConfigurationError( + f"Doris MySQL TLS validation failed for {context}" + ) from None access_error = _mysql_access_error(exc, operation="schema-discovery", context=context) if access_error is not None: raise access_error from None diff --git a/src/ray_doris/_readers.py b/src/ray_doris/_readers.py index 0e34c79..1a40b71 100644 --- a/src/ray_doris/_readers.py +++ b/src/ray_doris/_readers.py @@ -10,16 +10,18 @@ import pyarrow as pa import pymysql -from ray_doris._errors import DorisReadError +from ray_doris._errors import DorisConfigurationError, DorisReadError from ray_doris._models import ( _ADBC_CONNECT_TIMEOUT_OPTION, DorisInputSplit, DorisReadConfig, ) from ray_doris._planner import ( + _contains_mysql_tls_error, _mysql_access_error, _mysql_connection_kwargs, _mysql_error_code, + _mysql_tls_is_configured, ) from ray_doris._schema import coerce_decimal from ray_doris._sql import build_select_sql @@ -177,9 +179,13 @@ def read_mysql( reached_eof=reached_eof, context=context, ) + except DorisConfigurationError: + raise except DorisReadError as exc: raise DorisReadError(f"failed to read Doris {context}: {exc}") from exc except pymysql.MySQLError as exc: + if _contains_mysql_tls_error(exc): + raise DorisReadError(f"Doris MySQL TLS validation failed for {context}") from None access_error = _mysql_access_error( exc, operation="split read", @@ -191,6 +197,12 @@ def read_mysql( f"failed to read Doris {context} through the " f"MySQL protocol (MySQL error {_mysql_error_code(exc)!r})" ) from None + except OSError: + if _mysql_tls_is_configured(config): + raise DorisReadError( + f"Doris MySQL TLS configuration is invalid for {context}" + ) from None + raise DorisReadError(f"failed to read Doris {context} through the MySQL protocol") from None except Exception: raise DorisReadError(f"failed to read Doris {context} through the MySQL protocol") from None @@ -218,7 +230,7 @@ def _flight_connection(config: DorisReadConfig) -> Any: raise _flight_import_error() from None db_kwargs = { DatabaseOptions.USERNAME.value: config.user, - DatabaseOptions.PASSWORD.value: config.password, + DatabaseOptions.PASSWORD.value: config.resolve_password(), _ADBC_CONNECT_TIMEOUT_OPTION: str(config.connect_timeout), } db_kwargs.update(config.adbc_options()) @@ -303,7 +315,7 @@ def read_flight( cursor.close() finally: connection.close() - except _FlightUnavailableError: + except (_FlightUnavailableError, DorisConfigurationError): raise except DorisReadError as exc: raise DorisReadError( diff --git a/src/ray_doris/datasource.py b/src/ray_doris/datasource.py index 2ff02eb..cbb96be 100644 --- a/src/ray_doris/datasource.py +++ b/src/ray_doris/datasource.py @@ -64,6 +64,7 @@ def __init__( flight_scheme: FlightScheme = "grpc", user: str = "root", password: str = "", + password_env: Optional[str] = None, columns: Optional[Sequence[str]] = None, filter: Optional[str] = None, transport: Transport = "mysql", @@ -71,6 +72,8 @@ def __init__( tablet_size: int = 1, batch_size: int = 10_000, connect_timeout: float = 10.0, + query_plan_timeout: Optional[float] = None, + http_ca_file: Optional[str] = None, client_kwargs: Optional[Mapping[str, Any]] = None, flight_options: Optional[Mapping[str, Any]] = None, ) -> None: @@ -88,6 +91,7 @@ def __init__( flight_scheme=flight_scheme, user=user, password=password, + password_env=password_env, columns=normalize_columns(columns), filter=normalize_filter(filter), transport=transport, @@ -95,6 +99,8 @@ def __init__( tablet_size=tablet_size, batch_size=batch_size, connect_timeout=connect_timeout, + query_plan_timeout=query_plan_timeout, + http_ca_file=http_ca_file, client_options=client_kwargs, flight_options=flight_options, ) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 230fde2..b9c09e5 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -9,6 +9,8 @@ import pytest import ray +READER_PASSWORD_ENV = "RAY_DORIS_IT_READER_PASSWORD" + @dataclass(frozen=True) class DorisITConfig: @@ -43,6 +45,17 @@ def minimal_reader_kwargs(self, *, transport: str = "mysql", **kwargs: object) - values.update(user=self.reader_user, password=self.reader_password) return values + def minimal_env_reader_kwargs( + self, + *, + transport: str = "mysql", + password_env: str = READER_PASSWORD_ENV, + **kwargs: object, + ) -> dict: + values = self.reader_kwargs(transport=transport, **kwargs) + values.update(user=self.reader_user, password="", password_env=password_env) + return values + def _config_from_environment() -> DorisITConfig: return DorisITConfig( @@ -192,9 +205,15 @@ def doris_config() -> Iterator[DorisITConfig]: @pytest.fixture(scope="session", autouse=True) -def ray_runtime() -> Iterator[None]: +def ray_runtime(doris_config: DorisITConfig) -> Iterator[None]: + previous_password = os.environ.get(READER_PASSWORD_ENV) + os.environ[READER_PASSWORD_ENV] = doris_config.reader_password ray.init(num_cpus=4, include_dashboard=False) try: yield finally: ray.shutdown() + if previous_password is None: + os.environ.pop(READER_PASSWORD_ENV, None) + else: + os.environ[READER_PASSWORD_ENV] = previous_password diff --git a/tests/integration/test_mysql_read.py b/tests/integration/test_mysql_read.py index a3b3597..c47a770 100644 --- a/tests/integration/test_mysql_read.py +++ b/tests/integration/test_mysql_read.py @@ -1,15 +1,32 @@ import gc +import traceback from datetime import datetime from decimal import Decimal import pymysql import pytest -from ray_doris import DorisConfigurationError, DorisDatasource, DorisPlanningError, read_doris +from ray_doris import ( + DorisAuthenticationError, + DorisConfigurationError, + DorisDatasource, + DorisPlanningError, + read_doris, +) pytestmark = pytest.mark.integration +def _assert_redacted_exception(exception: BaseException, caplog, sentinel: str) -> None: + rendered_traceback = "".join( + traceback.format_exception(type(exception), exception, exception.__traceback__) + ) + assert sentinel not in str(exception) + assert sentinel not in rendered_traceback + assert sentinel not in caplog.text + assert exception.__cause__ is None + + def test_mysql_reads_projected_filtered_rows_in_bounded_batches(doris_config) -> None: dataset = read_doris( **doris_config.reader_kwargs( @@ -110,6 +127,72 @@ def test_minimum_select_privilege_reads_through_public_entrypoint(doris_config) assert sorted(row["id"] for row in rows) == [1, 2] +def test_environment_password_reads_on_driver_and_ray_workers(doris_config) -> None: + rows = read_doris( + **doris_config.minimal_env_reader_kwargs( + columns=["id"], + filter="id <= 2", + on_query_plan_error="error", + ) + ).take_all() + assert sorted(row["id"] for row in rows) == [1, 2] + + +def test_missing_environment_password_fails_before_driver_network( + doris_config, monkeypatch, caplog +) -> None: + variable = "RAY_DORIS_IT_MISSING_PASSWORD_SENTINEL" + monkeypatch.delenv(variable, raising=False) + datasource = DorisDatasource( + **doris_config.minimal_env_reader_kwargs( + password_env=variable, + on_query_plan_error="error", + ) + ) + with pytest.raises( + DorisConfigurationError, match="environment variable is unavailable" + ) as captured: + datasource.get_read_tasks(parallelism=1) + _assert_redacted_exception(captured.value, caplog, variable) + + +def test_environment_password_is_resolved_again_when_split_starts( + doris_config, monkeypatch, caplog +) -> None: + variable = "RAY_DORIS_IT_RETRY_PASSWORD_SENTINEL" + monkeypatch.setenv(variable, doris_config.reader_password) + datasource = DorisDatasource( + **doris_config.minimal_env_reader_kwargs( + password_env=variable, + columns=["id"], + tablet_size=4, + on_query_plan_error="error", + ) + ) + task = datasource.get_read_tasks(parallelism=1)[0] + monkeypatch.delenv(variable) + with pytest.raises( + DorisConfigurationError, match="environment variable is unavailable" + ) as captured: + list(task()) + _assert_redacted_exception(captured.value, caplog, variable) + + +def test_incorrect_environment_password_fails_closed(doris_config, monkeypatch, caplog) -> None: + variable = "RAY_DORIS_IT_WRONG_PASSWORD" + sentinel = "wrong-password-secret-sentinel" + monkeypatch.setenv(variable, sentinel) + datasource = DorisDatasource( + **doris_config.minimal_env_reader_kwargs( + password_env=variable, + on_query_plan_error="error", + ) + ) + with pytest.raises(DorisAuthenticationError, match="schema-discovery") as captured: + datasource.get_read_tasks(parallelism=1) + _assert_redacted_exception(captured.value, caplog, sentinel) + + def test_mysql_consumer_close_does_not_drain_active_unbuffered_result( doris_config, monkeypatch ) -> None: diff --git a/tests/slow_integration/_cluster.py b/tests/slow_integration/_cluster.py index 20997d7..01700ab 100644 --- a/tests/slow_integration/_cluster.py +++ b/tests/slow_integration/_cluster.py @@ -14,6 +14,8 @@ DISTRIBUTED_TABLE = "distributed_records" REPLICATED_TABLE = "replicated_records" FLIGHT_PROXY_BACKENDS = ("be-1", "be-2", "be-3") +READER_USER = "ray_doris_reader" +READER_PASSWORD_ENV = "RAY_DORIS_READER_PASSWORD" def _positive_env_int(name: str, default: int, maximum: int) -> int: @@ -89,7 +91,7 @@ def reader_kwargs( self, *, table: str, - transport: str = "flight", + transport: str = "mysql", **kwargs: Any, ) -> dict[str, Any]: values: dict[str, Any] = { @@ -100,11 +102,13 @@ def reader_kwargs( "http_scheme": "https", "flight_port": self.flight_port, "flight_scheme": "grpc", - "user": "root", - "password": "", + "user": READER_USER, + "password_env": READER_PASSWORD_ENV, "transport": transport, "on_query_plan_error": "error", "connect_timeout": 30.0, + "query_plan_timeout": 30.0, + "http_ca_file": self.tls_ca, "client_kwargs": { "ssl": { "ca": self.tls_ca, @@ -138,6 +142,32 @@ def mysql_connection(config: SlowITConfig): ) +def _reader_password() -> str: + password = os.environ.get(READER_PASSWORD_ENV) + if not password: + raise RuntimeError(f"{READER_PASSWORD_ENV} must be set to a non-empty test credential") + if not password.isascii() or not password.isalnum(): + raise RuntimeError(f"{READER_PASSWORD_ENV} must contain only ASCII letters and digits") + return password + + +def reader_mysql_connection(config: SlowITConfig): + return pymysql.connect( + host=config.host, + port=config.mysql_port, + user=READER_USER, + password=_reader_password(), + database=config.database, + charset="utf8mb4", + autocommit=True, + cursorclass=pymysql.cursors.DictCursor, + connect_timeout=30, + read_timeout=180, + write_timeout=180, + ssl={"ca": config.tls_ca, "check_hostname": True}, + ) + + def query_rows(config: SlowITConfig, sql: str) -> list[dict[str, Any]]: connection = mysql_connection(config) try: @@ -264,6 +294,16 @@ def setup_tables(config: SlowITConfig) -> None: PROPERTIES ("replication_num" = "1") """, ) + reader_password = _reader_password() + execute(config, f"DROP USER IF EXISTS '{READER_USER}'@'%'") + execute( + config, + f"CREATE USER '{READER_USER}'@'%' IDENTIFIED BY '{reader_password}'", + ) + execute( + config, + f"GRANT SELECT_PRIV ON internal.{config.database}.* TO '{READER_USER}'@'%'", + ) execute( config, f""" diff --git a/tests/slow_integration/docker-compose.yml b/tests/slow_integration/docker-compose.yml index b4b2f34..03874ff 100644 --- a/tests/slow_integration/docker-compose.yml +++ b/tests/slow_integration/docker-compose.yml @@ -15,7 +15,11 @@ x-ray-common: &ray-common DORIS_FLIGHT_PORT: "18070" DORIS_FLIGHT_PROXY_STATS_URL: "http://flight-proxy:8404/stats;csv" DORIS_TLS_CA: /tls/ca.pem + RAY_DORIS_READER_PASSWORD: "${RAY_DORIS_READER_PASSWORD:?run.sh must provide the reader password}" RAY_DORIS_ROW_COUNT: "${RAY_DORIS_ROW_COUNT:-10000}" + RAY_DORIS_SLOW_COMMIT_SHA: "${RAY_DORIS_SLOW_COMMIT_SHA:?run.sh must provide the commit SHA}" + RAY_DORIS_SLOW_PROFILE: "${RAY_DORIS_SLOW_PROFILE:?run.sh must provide the profile}" + RAY_DORIS_SLOW_RUN_ID: "${RAY_DORIS_SLOW_RUN_ID:?run.sh must provide the run ID}" RAY_DORIS_STRESS_SECONDS: "${RAY_DORIS_STRESS_SECONDS:-5}" RAY_WORKER_IPS: 172.31.128.11,172.31.128.12,172.31.128.13 RAY_USAGE_STATS_ENABLED: "0" @@ -163,6 +167,7 @@ services: networks: doris: ipv4_address: 172.31.128.6 + aliases: [doris-ingress] volumes: - ./docker/haproxy.cfg:/usr/local/etc/haproxy/haproxy.cfg:ro - tls:/tls:ro diff --git a/tests/slow_integration/docker/generate-tls.sh b/tests/slow_integration/docker/generate-tls.sh index d7a8962..3f6a719 100755 --- a/tests/slow_integration/docker/generate-tls.sh +++ b/tests/slow_integration/docker/generate-tls.sh @@ -43,6 +43,18 @@ openssl x509 \ -extfile "${target}/fe.ext" \ -out "${target}/fe.pem" +openssl req \ + -x509 \ + -newkey rsa:2048 \ + -nodes \ + -sha256 \ + -days 2 \ + -keyout "${target}/wrong-ca.key" \ + -out "${target}/wrong-ca.pem" \ + -subj "/CN=ray-doris untrusted test CA" \ + -addext "basicConstraints=critical,CA:TRUE" \ + -addext "keyUsage=critical,keyCertSign,cRLSign" + openssl pkcs12 -export -name doris_ssl_certificate -inkey "${target}/fe.key" \ -in "${target}/fe.pem" -certfile "${target}/ca.pem" -out "${target}/fe.p12" \ -passout pass:doris @@ -53,4 +65,5 @@ chmod 0644 \ "${target}/ca.pem" \ "${target}/fe.p12" \ "${target}/ingress.pem" \ - "${target}/mysql-ca.p12" + "${target}/mysql-ca.p12" \ + "${target}/wrong-ca.pem" diff --git a/tests/slow_integration/run.sh b/tests/slow_integration/run.sh index 22e4a48..7e8d7c4 100755 --- a/tests/slow_integration/run.sh +++ b/tests/slow_integration/run.sh @@ -6,9 +6,19 @@ compose_file="${script_dir}/docker-compose.yml" log_dir=${RAY_DORIS_SLOW_LOG_DIR:-/tmp/ray-doris-slow-it-logs} pytest_log="${log_dir}/pytest.log" compose_log="${log_dir}/compose.log" +result_file="${log_dir}/slow-result.json" compose=(docker compose -f "${compose_file}") pytest_pid= profile=${RAY_DORIS_SLOW_PROFILE:-full} +RAY_DORIS_SLOW_COMMIT_SHA=${GITHUB_SHA:-$(git -C "${script_dir}/../.." rev-parse HEAD)} +RAY_DORIS_SLOW_RUN_ID=${GITHUB_RUN_ID:-1} +RAY_DORIS_SLOW_PROFILE=${profile} +export RAY_DORIS_SLOW_COMMIT_SHA RAY_DORIS_SLOW_PROFILE RAY_DORIS_SLOW_RUN_ID + +if [[ -z "${RAY_DORIS_READER_PASSWORD:-}" ]]; then + RAY_DORIS_READER_PASSWORD=$(openssl rand -hex 24) + export RAY_DORIS_READER_PASSWORD +fi if [[ "${profile}" != "full" && "${profile}" != "core" ]]; then echo "RAY_DORIS_SLOW_PROFILE must be 'full' or 'core'." >&2 @@ -92,7 +102,7 @@ if [[ "${profile}" == "core" ]]; then "${compose[@]}" exec -T ray-head \ python -m pytest \ -m slow_integration \ - tests/slow_integration/test_distributed_cluster.py::test_flight_read_executes_on_all_ray_workers \ + tests/slow_integration/test_distributed_cluster.py::test_mysql_read_executes_on_all_ray_workers \ -vv -s 2>&1 | tee -a "${pytest_log}" exit 0 fi @@ -114,3 +124,7 @@ wait_for_marker /state/be-failure-ready 600 wait "${pytest_pid}" pytest_pid= + +"${compose[@]}" exec -T ray-head \ + python tests/slow_integration/write_result.py +"${compose[@]}" cp ray-head:/state/slow-result.json "${result_file}" diff --git a/tests/slow_integration/test_distributed_cluster.py b/tests/slow_integration/test_distributed_cluster.py index 6f68ff4..29f3174 100644 --- a/tests/slow_integration/test_distributed_cluster.py +++ b/tests/slow_integration/test_distributed_cluster.py @@ -3,6 +3,7 @@ import os import ssl import time +import traceback from collections.abc import Callable, Iterable, Iterator from functools import partial from pathlib import Path @@ -18,17 +19,19 @@ SlowITConfig, alive_backends, flight_proxy_backend_stats, - mysql_connection, query_rows, + reader_mysql_connection, replica_distribution, wait_for_backend_count, - wait_for_flight_proxy_backend_count, ) from ray.data.aggregate import Count, Max, Min, Sum from ray.data.datasource import ReadTask from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy -from ray_doris import DorisDatasource, read_doris +from ray_doris import DorisConfigurationError, DorisDatasource, read_doris +from ray_doris._models import DorisReadConfig +from ray_doris._planner import QueryPlanClient +from ray_doris._sql import build_select_sql, parse_table pytestmark = [ pytest.mark.slow_integration, @@ -40,6 +43,16 @@ BE_FAILURE_MARKER = "/state/be-failure-ready" +def _assert_redacted_exception(exception: BaseException, caplog, sentinel: str) -> None: + rendered_traceback = "".join( + traceback.format_exception(type(exception), exception, exception.__traceback__) + ) + assert sentinel not in str(exception) + assert sentinel not in rendered_traceback + assert sentinel not in caplog.text + assert exception.__cause__ is None + + def _alive_ray_nodes() -> list[dict[str, Any]]: return [node for node in ray.nodes() if node["Alive"]] @@ -196,7 +209,7 @@ def test_cluster_topology_and_replica_distribution( def test_https_gateway_mysql_tls_and_explicit_flight( slow_config: SlowITConfig, ) -> None: - connection = mysql_connection(slow_config) + connection = reader_mysql_connection(slow_config) try: assert isinstance(connection._sock, ssl.SSLSocket) assert connection._sock.cipher() is not None @@ -229,14 +242,77 @@ def test_https_gateway_mysql_tls_and_explicit_flight( ) -def test_flight_read_executes_on_all_ray_workers( +def test_tls_rejects_wrong_ca_hostname_and_plaintext_endpoint( slow_config: SlowITConfig, + caplog, +) -> None: + sentinel = os.environ["RAY_DORIS_READER_PASSWORD"] + wrong_ca_kwargs = slow_config.reader_kwargs( + table=slow_config.distributed_table, + http_ca_file="/tls/wrong-ca.pem", + client_kwargs={ + "ssl": {"ca": "/tls/wrong-ca.pem", "check_hostname": True}, + "read_timeout": 180, + "write_timeout": 180, + }, + ) + with pytest.raises(DorisConfigurationError, match="MySQL TLS validation failed") as captured: + DorisDatasource(**wrong_ca_kwargs).get_read_tasks(parallelism=1) + _assert_redacted_exception(captured.value, caplog, sentinel) + + wrong_hostname_kwargs = slow_config.reader_kwargs( + table=slow_config.distributed_table, + host="doris-ingress", + ) + with pytest.raises(DorisConfigurationError, match="MySQL TLS validation failed") as captured: + DorisDatasource(**wrong_hostname_kwargs).get_read_tasks(parallelism=1) + _assert_redacted_exception(captured.value, caplog, sentinel) + + query_plan_config = DorisReadConfig.from_options( + table=parse_table(f"{slow_config.database}.{slow_config.distributed_table}"), + host="doris-ingress", + mysql_port=slow_config.mysql_port, + http_port=slow_config.https_port, + http_scheme="https", + http_ca_file=slow_config.tls_ca, + user="ray_doris_reader", + password_env="RAY_DORIS_READER_PASSWORD", + on_query_plan_error="error", + ) + query_plan_sql = build_select_sql(query_plan_config.table, None, None, None) + with pytest.raises( + DorisConfigurationError, match="query-plan TLS validation failed" + ) as captured: + QueryPlanClient(query_plan_config).fetch_tablet_ids(query_plan_sql) + _assert_redacted_exception(captured.value, caplog, sentinel) + + plaintext_config = DorisReadConfig.from_options( + table=query_plan_config.table, + host="fe", + http_port=8030, + http_scheme="https", + http_ca_file=slow_config.tls_ca, + user=query_plan_config.user, + password_env="RAY_DORIS_READER_PASSWORD", + on_query_plan_error="error", + ) + with pytest.raises( + DorisConfigurationError, match="query-plan TLS validation failed" + ) as captured: + QueryPlanClient(plaintext_config).fetch_tablet_ids(query_plan_sql) + _assert_redacted_exception(captured.value, caplog, sentinel) + + +def _assert_read_executes_on_all_ray_workers( + slow_config: SlowITConfig, + *, + transport: str, ) -> None: - proxy_before = flight_proxy_backend_stats(slow_config) + proxy_before = flight_proxy_backend_stats(slow_config) if transport == "flight" else None datasource = InstrumentedDorisDatasource( **slow_config.reader_kwargs( table=slow_config.distributed_table, - transport="flight", + transport=transport, columns=["id", WORKER_COLUMN], tablet_size=1, batch_size=5_000, @@ -270,13 +346,26 @@ def test_flight_read_executes_on_all_ray_workers( assert row_count == slow_config.row_count assert id_sum == slow_config.expected_id_sum - proxy_after = flight_proxy_backend_stats(slow_config) - for backend in proxy_before: - previous = proxy_before[backend] - current = proxy_after[backend] - assert current.status == "UP" - assert current.total_sessions > previous.total_sessions - assert current.bytes_in + current.bytes_out > previous.bytes_in + previous.bytes_out + if proxy_before is not None: + proxy_after = flight_proxy_backend_stats(slow_config) + for backend in proxy_before: + previous = proxy_before[backend] + current = proxy_after[backend] + assert current.status == "UP" + assert current.total_sessions > previous.total_sessions + assert current.bytes_in + current.bytes_out > previous.bytes_in + previous.bytes_out + + +def test_mysql_read_executes_on_all_ray_workers( + slow_config: SlowITConfig, +) -> None: + _assert_read_executes_on_all_ray_workers(slow_config, transport="mysql") + + +def test_flight_read_executes_on_all_ray_workers( + slow_config: SlowITConfig, +) -> None: + _assert_read_executes_on_all_ray_workers(slow_config, transport="flight") def test_repeated_flight_reads( @@ -331,7 +420,7 @@ def test_worker_retry_and_backend_failover( datasource = InstrumentedDorisDatasource( **slow_config.reader_kwargs( table=slow_config.replicated_table, - transport="flight", + transport="mysql", columns=["id", WORKER_COLUMN], filter=f"id < {retry_rows}", tablet_size=BUCKET_COUNT, @@ -381,17 +470,10 @@ def test_worker_retry_and_backend_failover( timeout_seconds=180, ) assert len(surviving_backends) == 2 - proxy_stats = wait_for_flight_proxy_backend_count( - slow_config, - 2, - timeout_seconds=180, - ) - assert proxy_stats["be-1"].status == "DOWN" - dataset = read_doris( **slow_config.reader_kwargs( table=slow_config.replicated_table, - transport="flight", + transport="mysql", columns=["id"], tablet_size=1, batch_size=10_000, diff --git a/tests/slow_integration/write_result.py b/tests/slow_integration/write_result.py new file mode 100644 index 0000000..f14cadf --- /dev/null +++ b/tests/slow_integration/write_result.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import json +import os +import platform +from pathlib import Path +from typing import Any + +import ray +from _cluster import SlowITConfig + + +def _required_environment(name: str) -> str: + value = os.environ.get(name) + if not value: + raise RuntimeError(f"{name} is required to write slow-suite evidence") + return value + + +def write_result(path: Path) -> None: + config = SlowITConfig.from_environment() + result: dict[str, Any] = { + "schema_version": 1, + "commit_sha": _required_environment("RAY_DORIS_SLOW_COMMIT_SHA"), + "workflow_run_id": int(_required_environment("RAY_DORIS_SLOW_RUN_ID")), + "profile": _required_environment("RAY_DORIS_SLOW_PROFILE"), + "status": "passed", + "transport": "mysql", + "endpoint_mode": "logical", + "doris_version": "4.0.6", + "ray_version": ray.__version__, + "python_version": platform.python_version(), + "initial_frontend_count": 1, + "initial_backend_count": 3, + "initial_ray_worker_count": len(config.worker_ips), + "row_count": config.row_count, + "scenarios": [ + "be_failure", + "logical_endpoint_tls", + "mysql_all_workers", + "ray_worker_retry", + ], + } + path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +if __name__ == "__main__": + write_result(Path("/state/slow-result.json")) diff --git a/tests/unit/test_api.py b/tests/unit/test_api.py index 3136554..b7a672c 100644 --- a/tests/unit/test_api.py +++ b/tests/unit/test_api.py @@ -13,8 +13,11 @@ def test_read_doris_uses_public_ray_entrypoint(monkeypatch) -> None: table="db.table", host="fe", http_scheme="https", + http_ca_file="ca.pem", flight_scheme="grpc+tls", connect_timeout=3.5, + query_plan_timeout=4.5, + password_env="RAY_DORIS_PASSWORD", concurrency=2, override_num_blocks=4, ray_remote_args={"num_cpus": 0.25}, @@ -24,6 +27,9 @@ def test_read_doris_uses_public_ray_entrypoint(monkeypatch) -> None: assert datasource.config.http_scheme == "https" assert datasource.config.flight_scheme == "grpc+tls" assert datasource.config.connect_timeout == 3.5 + assert datasource.config.query_plan_timeout == 4.5 + assert datasource.config.http_ca_file == "ca.pem" + assert datasource.config.password_env == "RAY_DORIS_PASSWORD" assert read_datasource.call_args.kwargs == { "concurrency": 2, "override_num_blocks": 4, diff --git a/tests/unit/test_datasource.py b/tests/unit/test_datasource.py index 7026822..429c5ab 100644 --- a/tests/unit/test_datasource.py +++ b/tests/unit/test_datasource.py @@ -74,6 +74,25 @@ def test_get_read_tasks_has_current_compatible_signature_and_schema(monkeypatch) assert restored.metadata == tasks[0].metadata datasource.get_read_tasks(2) discover.assert_called_once() + + +def test_environment_password_never_enters_datasource_or_read_task_pickle(monkeypatch) -> None: + secret = "datasource-password-secret-sentinel" + monkeypatch.setenv("RAY_DORIS_TASK_PASSWORD", secret) + schema = pa.schema([("id", pa.int64())]) + discover = Mock(return_value=DorisPlanningSnapshot(schema=schema, tablet_ids=(7,))) + monkeypatch.setattr( + "ray_doris.datasource.DorisPlanner.discover", + discover, + ) + datasource = DorisDatasource( + table="db.table", + host="fe", + password_env="RAY_DORIS_TASK_PASSWORD", + ) + task = datasource.get_read_tasks(1)[0] + assert secret.encode() not in cloudpickle.dumps(datasource) + assert secret.encode() not in cloudpickle.dumps(task) restored_datasource = cloudpickle.loads(cloudpickle.dumps(datasource)) assert restored_datasource.get_read_tasks(2)[0].schema == schema discover.assert_called_once() diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py index f87b471..45c93bf 100644 --- a/tests/unit/test_models.py +++ b/tests/unit/test_models.py @@ -1,4 +1,5 @@ import math +import traceback import pytest from ray import cloudpickle @@ -14,10 +15,23 @@ def make_config(**kwargs: object) -> DorisReadConfig: return DorisReadConfig.from_options(**values) +def _assert_redacted_exception(exception: BaseException, caplog, sentinel: str) -> None: + rendered_traceback = "".join( + traceback.format_exception(type(exception), exception, exception.__traceback__) + ) + assert sentinel not in str(exception) + assert sentinel not in rendered_traceback + assert sentinel not in caplog.text + assert exception.__cause__ is None + + def test_config_defaults() -> None: config = make_config() assert config.transport == "mysql" assert config.connect_timeout == 10.0 + assert config.effective_query_plan_timeout == 10.0 + assert config.password_env is None + assert config.http_ca_file is None def test_config_repr_redacts_password_and_option_values() -> None: @@ -36,6 +50,20 @@ def test_config_repr_redacts_password_and_option_values() -> None: assert "" in rendered +def test_config_repr_redacts_environment_name_and_http_ca_path() -> None: + rendered = repr( + make_config( + password_env="RAY_DORIS_PASSWORD_SECRET_NAME", + http_scheme="https", + http_ca_file="/private/tls/sensitive-ca-name.pem", + ) + ) + assert "RAY_DORIS_PASSWORD_SECRET_NAME" not in rendered + assert "sensitive-ca-name.pem" not in rendered + assert "password_env=" in rendered + assert "http_ca_file=" in rendered + + def test_config_repr_distinguishes_missing_filter_without_exposing_values() -> None: assert "filter=None" in repr(make_config()) @@ -53,6 +81,34 @@ def test_config_is_cloudpickle_serializable_and_freezes_options() -> None: assert restored.mysql_options()["ssl"]["ca"] == "original.pem" +def test_config_resolves_environment_password_without_serializing_value(monkeypatch) -> None: + secret = "resolved-password-secret-sentinel" + monkeypatch.setenv("RAY_DORIS_TEST_PASSWORD", secret) + config = make_config(password_env="RAY_DORIS_TEST_PASSWORD") + assert config.resolve_password() == secret + assert secret.encode() not in cloudpickle.dumps(config) + + +def test_config_environment_password_preserves_empty_value(monkeypatch) -> None: + monkeypatch.setenv("RAY_DORIS_TEST_PASSWORD", "") + assert make_config(password_env="RAY_DORIS_TEST_PASSWORD").resolve_password() == "" + + +def test_config_missing_environment_password_is_redacted(monkeypatch, caplog) -> None: + sentinel = "RAY_DORIS_MISSING_PASSWORD_SENTINEL" + monkeypatch.delenv(sentinel, raising=False) + with pytest.raises( + DorisConfigurationError, match="environment variable is unavailable" + ) as captured: + make_config(password_env=sentinel).resolve_password() + _assert_redacted_exception(captured.value, caplog, sentinel) + + +def test_config_query_plan_timeout_overrides_legacy_default() -> None: + config = make_config(connect_timeout=3.0, query_plan_timeout=7.5) + assert config.effective_query_plan_timeout == 7.5 + + @pytest.mark.parametrize( "timeout", [math.nan, math.inf, -math.inf, 31_536_001, 10**1000], @@ -65,6 +121,28 @@ def test_config_rejects_invalid_connect_timeout(timeout: float) -> None: ) +@pytest.mark.parametrize("timeout", [True, "30", 0, math.nan, math.inf, 31_536_001]) +def test_config_rejects_invalid_query_plan_timeout(timeout: object) -> None: + with pytest.raises(DorisConfigurationError, match="query_plan_timeout"): + make_config(query_plan_timeout=timeout) + + +@pytest.mark.parametrize("name", ["", "1PASSWORD", "HAS SPACE", "HAS-DASH", b"PASSWORD"]) +def test_config_rejects_invalid_password_environment_name(name: object) -> None: + with pytest.raises(DorisConfigurationError, match="password_env"): + make_config(password_env=name) + + +def test_config_rejects_literal_and_environment_password_together() -> None: + with pytest.raises(DorisConfigurationError, match="mutually exclusive"): + make_config(password="literal", password_env="RAY_DORIS_PASSWORD") + + +def test_config_requires_https_for_http_ca_file() -> None: + with pytest.raises(DorisConfigurationError, match="requires http_scheme='https'"): + make_config(http_ca_file="ca.pem") + + @pytest.mark.parametrize("value", ["30", 0, math.nan, math.inf]) def test_config_rejects_invalid_mysql_execution_timeout(value: object) -> None: with pytest.raises(DorisConfigurationError, match="read_timeout"): @@ -98,6 +176,7 @@ def __deepcopy__(self, memo): {"connect_timeout": 0}, {"transport": "unknown"}, {"http_scheme": "ftp"}, + {"http_scheme": "https", "http_ca_file": ""}, {"flight_scheme": "https"}, {"on_query_plan_error": "ignore"}, {"batch_size": "10"}, diff --git a/tests/unit/test_planner.py b/tests/unit/test_planner.py index 4883950..16c20fd 100644 --- a/tests/unit/test_planner.py +++ b/tests/unit/test_planner.py @@ -2,6 +2,7 @@ import io import json import ssl +import traceback import urllib.error from unittest.mock import Mock @@ -27,6 +28,16 @@ def make_config(**kwargs: object) -> DorisReadConfig: return DorisReadConfig.from_options(**values) +def _assert_redacted_exception(exception: BaseException, caplog, sentinel: str) -> None: + rendered_traceback = "".join( + traceback.format_exception(type(exception), exception, exception.__traceback__) + ) + assert sentinel not in str(exception) + assert sentinel not in rendered_traceback + assert sentinel not in caplog.text + assert exception.__cause__ is None + + def success_payload(partitions: object) -> object: return { "code": 0, @@ -69,6 +80,7 @@ def test_query_plan_http_client_sends_post_auth_scheme_and_timeout(monkeypatch) password="secret", http_scheme="https", connect_timeout=3.5, + query_plan_timeout=4.5, ) assert QueryPlanClient(config).fetch_tablet_ids("SELECT * FROM `db`.`table`") == (9,) request = open_request.call_args.args[0] @@ -76,7 +88,52 @@ def test_query_plan_http_client_sends_post_auth_scheme_and_timeout(monkeypatch) assert request.method == "POST" assert request.get_header("Authorization") == "Basic cmVhZGVyOnNlY3JldA==" assert request.data == b'{"sql": "SELECT * FROM `db`.`table`"}' - assert open_request.call_args.kwargs["timeout"] == 3.5 + assert open_request.call_args.kwargs["timeout"] == 4.5 + assert isinstance(open_request.call_args.kwargs["ssl_context"], ssl.SSLContext) + + +def test_query_plan_resolves_environment_password_for_each_request(monkeypatch) -> None: + response = FakeResponse(json.dumps(success_payload({"9": {}})).encode("utf-8")) + open_request = Mock(return_value=response) + monkeypatch.setattr(_planner, "_open_query_plan_request", open_request) + monkeypatch.setenv("RAY_DORIS_PLANNER_PASSWORD", "first-password") + config = make_config(password_env="RAY_DORIS_PLANNER_PASSWORD") + QueryPlanClient(config).fetch_tablet_ids("SELECT 1") + first_authorization = open_request.call_args.args[0].get_header("Authorization") + monkeypatch.setenv("RAY_DORIS_PLANNER_PASSWORD", "second-password") + QueryPlanClient(config).fetch_tablet_ids("SELECT 1") + second_authorization = open_request.call_args.args[0].get_header("Authorization") + assert first_authorization != second_authorization + + +def test_query_plan_missing_environment_password_fails_before_network(monkeypatch, caplog) -> None: + sentinel = "RAY_DORIS_QUERY_PLAN_MISSING_PASSWORD_SENTINEL" + open_request = Mock() + monkeypatch.setattr(_planner, "_open_query_plan_request", open_request) + monkeypatch.delenv(sentinel, raising=False) + with pytest.raises( + DorisConfigurationError, match="environment variable is unavailable" + ) as captured: + QueryPlanClient(make_config(password_env=sentinel)).fetch_tablet_ids("SELECT 1") + _assert_redacted_exception(captured.value, caplog, sentinel) + open_request.assert_not_called() + + +def test_query_plan_custom_ca_failure_is_redacted_before_network(monkeypatch, caplog) -> None: + sentinel = "query-plan-ca-path-secret-sentinel" + open_request = Mock() + monkeypatch.setattr(_planner, "_open_query_plan_request", open_request) + monkeypatch.setattr( + _planner.ssl, + "create_default_context", + Mock(side_effect=OSError(sentinel)), + ) + with pytest.raises(DorisConfigurationError, match="TLS configuration is invalid") as captured: + QueryPlanClient( + make_config(http_scheme="https", http_ca_file="private-ca.pem") + ).fetch_tablet_ids("SELECT 1") + _assert_redacted_exception(captured.value, caplog, sentinel) + open_request.assert_not_called() def test_query_plan_http_client_classifies_transport_and_invalid_json(monkeypatch) -> None: @@ -367,6 +424,30 @@ def test_describe_schema_uses_managed_connection_and_fetchmany(monkeypatch) -> N assert connect.call_args.kwargs["read_timeout"] == 30 +def test_describe_schema_resolves_environment_password(monkeypatch) -> None: + cursor = DescribeCursor() + connect = Mock(return_value=DescribeConnection(cursor)) + monkeypatch.setattr(_planner.pymysql, "connect", connect) + monkeypatch.setenv("RAY_DORIS_DESCRIBE_PASSWORD", "resolved-password") + DorisPlanner(make_config(password_env="RAY_DORIS_DESCRIBE_PASSWORD"))._describe_schema() + assert connect.call_args.kwargs["password"] == "resolved-password" + + +def test_describe_schema_missing_environment_password_fails_before_network( + monkeypatch, caplog +) -> None: + sentinel = "RAY_DORIS_DESCRIBE_MISSING_PASSWORD_SENTINEL" + connect = Mock() + monkeypatch.setattr(_planner.pymysql, "connect", connect) + monkeypatch.delenv(sentinel, raising=False) + with pytest.raises( + DorisConfigurationError, match="environment variable is unavailable" + ) as captured: + DorisPlanner(make_config(password_env=sentinel))._describe_schema() + _assert_redacted_exception(captured.value, caplog, sentinel) + connect.assert_not_called() + + def test_group_tablets_rejects_invalid_parallelism() -> None: for parallelism in (0, True): with pytest.raises(DorisConfigurationError, match="parallelism"): @@ -406,3 +487,37 @@ def test_describe_schema_classifies_mysql_errors_with_table_context( DorisPlanner(make_config())._describe_schema() assert "sensitive server detail" not in str(captured.value) assert captured.value.__cause__ is None + + +def test_describe_schema_classifies_mysql_tls_error_without_exposing_driver_text( + monkeypatch, + caplog, +) -> None: + sentinel = "driver-tls-secret-sentinel" + monkeypatch.setattr( + _planner.pymysql, + "connect", + Mock( + side_effect=pymysql.err.OperationalError( + 2003, + f"[SSL: CERTIFICATE_VERIFY_FAILED] {sentinel}", + ) + ), + ) + with pytest.raises(DorisConfigurationError, match="MySQL TLS validation failed") as captured: + DorisPlanner(make_config())._describe_schema() + _assert_redacted_exception(captured.value, caplog, sentinel) + + +def test_describe_schema_redacts_mysql_ca_setup_error(monkeypatch, caplog) -> None: + sentinel = "mysql-ca-path-secret-sentinel" + monkeypatch.setattr( + _planner.pymysql, + "connect", + Mock(side_effect=FileNotFoundError(sentinel)), + ) + with pytest.raises(DorisConfigurationError, match="TLS configuration is invalid") as captured: + DorisPlanner( + make_config(client_options={"ssl": {"ca": "private-ca.pem"}}) + )._describe_schema() + _assert_redacted_exception(captured.value, caplog, sentinel) diff --git a/tests/unit/test_readers.py b/tests/unit/test_readers.py index 0c12540..900ee5f 100644 --- a/tests/unit/test_readers.py +++ b/tests/unit/test_readers.py @@ -1,4 +1,5 @@ import sys +import traceback from datetime import datetime from decimal import Decimal from types import ModuleType @@ -10,6 +11,7 @@ from ray_doris import _readers from ray_doris._errors import ( DorisAuthenticationError, + DorisConfigurationError, DorisPermissionError, DorisReadError, ) @@ -28,6 +30,16 @@ def make_config(**kwargs: object) -> DorisReadConfig: return DorisReadConfig.from_options(**values) +def _assert_redacted_exception(exception: BaseException, caplog, sentinel: str) -> None: + rendered_traceback = "".join( + traceback.format_exception(type(exception), exception, exception.__traceback__) + ) + assert sentinel not in str(exception) + assert sentinel not in rendered_traceback + assert sentinel not in caplog.text + assert exception.__cause__ is None + + class FakeResult: def __init__(self) -> None: self.connection = None @@ -104,6 +116,47 @@ def test_mysql_reader_streams_fetchmany_batches_and_closes_resources(monkeypatch assert cursor.close_calls == connection.close_calls == 1 +def test_mysql_reader_resolves_environment_password_for_each_attempt(monkeypatch) -> None: + schema = pa.schema( + [("id", pa.int64()), ("amount", pa.decimal128(10, 2)), ("created_at", pa.timestamp("us"))] + ) + connect = Mock( + side_effect=[ + FakeConnection(FakeCursor()), + FakeConnection(FakeCursor()), + ] + ) + monkeypatch.setattr(_readers.pymysql, "connect", connect) + config = make_config(password_env="RAY_DORIS_WORKER_PASSWORD") + monkeypatch.setenv("RAY_DORIS_WORKER_PASSWORD", "first-password") + list(_readers.read_mysql(config, DorisInputSplit(None), schema)) + monkeypatch.setenv("RAY_DORIS_WORKER_PASSWORD", "second-password") + list(_readers.read_mysql(config, DorisInputSplit(None), schema)) + assert connect.call_args_list[0].kwargs["password"] == "first-password" + assert connect.call_args_list[1].kwargs["password"] == "second-password" + + +def test_mysql_reader_missing_environment_password_fails_before_network( + monkeypatch, caplog +) -> None: + sentinel = "RAY_DORIS_READER_MISSING_PASSWORD_SENTINEL" + connect = Mock() + monkeypatch.setattr(_readers.pymysql, "connect", connect) + monkeypatch.delenv(sentinel, raising=False) + with pytest.raises( + DorisConfigurationError, match="environment variable is unavailable" + ) as captured: + list( + _readers.read_mysql( + make_config(password_env=sentinel), + DorisInputSplit(None), + pa.schema([("id", pa.int64())]), + ) + ) + _assert_redacted_exception(captured.value, caplog, sentinel) + connect.assert_not_called() + + def test_mysql_reader_abort_closes_connection_without_draining_cursor(monkeypatch) -> None: events = [] cursor = FakeCursor(events) @@ -198,6 +251,49 @@ def test_mysql_reader_preserves_access_error(monkeypatch, code, error_type) -> N assert captured.value.__cause__ is None +def test_mysql_reader_classifies_tls_error_without_exposing_driver_text( + monkeypatch, caplog +) -> None: + sentinel = "worker-tls-secret-sentinel" + monkeypatch.setattr( + _readers.pymysql, + "connect", + Mock( + side_effect=_readers.pymysql.err.OperationalError( + 2003, + f"[SSL: CERTIFICATE_VERIFY_FAILED] {sentinel}", + ) + ), + ) + with pytest.raises(DorisReadError, match="MySQL TLS validation failed") as captured: + list( + _readers.read_mysql( + make_config(), + DorisInputSplit(None), + pa.schema([("id", pa.int64())]), + ) + ) + _assert_redacted_exception(captured.value, caplog, sentinel) + + +def test_mysql_reader_redacts_ca_setup_error(monkeypatch, caplog) -> None: + sentinel = "worker-ca-path-secret-sentinel" + monkeypatch.setattr( + _readers.pymysql, + "connect", + Mock(side_effect=FileNotFoundError(sentinel)), + ) + with pytest.raises(DorisReadError, match="TLS configuration is invalid") as captured: + list( + _readers.read_mysql( + make_config(client_options={"ssl": {"ca": "private-ca.pem"}}), + DorisInputSplit(None), + pa.schema([("id", pa.int64())]), + ) + ) + _assert_redacted_exception(captured.value, caplog, sentinel) + + def test_boolean_normalization_accepts_only_doris_zero_and_one() -> None: assert _readers._normalize_values([0, 1, None, True], pa.bool_()) == [ False, @@ -536,9 +632,11 @@ class DatabaseOptions: monkeypatch.setitem(sys.modules, "adbc_driver_manager", manager) timeout_key = "adbc.flight.sql.rpc.timeout_seconds.query" + monkeypatch.setenv("RAY_DORIS_FLIGHT_PASSWORD", "resolved-flight-password") _readers._flight_connection( make_config( flight_scheme="grpc+tls", + password_env="RAY_DORIS_FLIGHT_PASSWORD", flight_options={timeout_key: "30"}, ) ) @@ -548,6 +646,7 @@ class DatabaseOptions: == "10.0" ) assert connect.call_args.kwargs["db_kwargs"][timeout_key] == "30" + assert connect.call_args.kwargs["db_kwargs"]["password"] == "resolved-flight-password" def test_adbc_status_classification_uses_structured_status_code(monkeypatch) -> None: diff --git a/tests/unit/test_release.py b/tests/unit/test_release.py index e4b4d71..0c23cab 100644 --- a/tests/unit/test_release.py +++ b/tests/unit/test_release.py @@ -1,8 +1,21 @@ +import json +import traceback from pathlib import Path from unittest.mock import Mock import pytest from tools import check_release +from tools.check_slow_result import verify_slow_result + + +def _assert_redacted_exception(exception: BaseException, caplog, sentinel: str) -> None: + rendered_traceback = "".join( + traceback.format_exception(type(exception), exception, exception.__traceback__) + ) + assert sentinel not in str(exception) + assert sentinel not in rendered_traceback + assert sentinel not in caplog.text + assert exception.__cause__ is None def write_pyproject(path: Path, version: str = "0.1.0a1") -> Path: @@ -64,3 +77,94 @@ def test_release_main_requires_github_tag_environment(monkeypatch) -> None: monkeypatch.delenv("GITHUB_SHA", raising=False) with pytest.raises(RuntimeError, match="GITHUB_REF_NAME"): check_release.main() + + +def write_slow_result(path: Path, **overrides: object) -> Path: + data = { + "schema_version": 1, + "commit_sha": "a" * 40, + "workflow_run_id": 42, + "profile": "full", + "status": "passed", + "transport": "mysql", + "endpoint_mode": "logical", + "doris_version": "4.0.6", + "ray_version": "2.55.1", + "python_version": "3.12.12", + "initial_frontend_count": 1, + "initial_backend_count": 3, + "initial_ray_worker_count": 3, + "row_count": 10_000, + "scenarios": [ + "be_failure", + "logical_endpoint_tls", + "mysql_all_workers", + "ray_worker_retry", + ], + } + data.update(overrides) + manifest = path / "slow-result.json" + manifest.write_text(json.dumps(data), encoding="utf-8") + return manifest + + +def test_slow_result_accepts_exact_full_profile_evidence(tmp_path) -> None: + verify_slow_result( + write_slow_result(tmp_path), + expected_commit="a" * 40, + expected_run_id=42, + ) + + +@pytest.mark.parametrize( + ("overrides", "message"), + [ + ({"commit_sha": "b" * 40}, "release commit"), + ({"profile": "core"}, "profile"), + ({"status": "failed"}, "status"), + ({"transport": "flight"}, "transport"), + ({"endpoint_mode": "multi-fe"}, "endpoint_mode"), + ({"schema_version": True}, "schema version"), + ({"ray_version": "2.56.1"}, "ray_version"), + ({"python_version": "3.11.9"}, "python_version"), + ({"initial_frontend_count": 2}, "initial_frontend_count"), + ({"initial_frontend_count": True}, "initial_frontend_count"), + ({"initial_backend_count": 2}, "initial_backend_count"), + ({"initial_ray_worker_count": 2}, "initial_ray_worker_count"), + ({"row_count": 0}, "row_count"), + ({"workflow_run_id": True}, "workflow_run_id"), + ({"scenarios": ["mysql_all_workers"]}, "scenarios"), + ( + { + "scenarios": [ + "be_failure", + "logical_endpoint_tls", + "mysql_all_workers", + "ray_worker_retry", + "ray_worker_retry", + ] + }, + "scenarios", + ), + ], +) +def test_slow_result_rejects_invalid_release_evidence( + tmp_path, + overrides: dict[str, object], + message: str, +) -> None: + with pytest.raises(RuntimeError, match=message): + verify_slow_result( + write_slow_result(tmp_path, **overrides), + expected_commit="a" * 40, + expected_run_id=42, + ) + + +def test_slow_result_rejects_unreadable_manifest_without_parser_details(tmp_path, caplog) -> None: + sentinel = "secret-json-sentinel" + manifest = tmp_path / "slow-result.json" + manifest.write_text("{" + sentinel, encoding="utf-8") + with pytest.raises(RuntimeError, match="manifest is unreadable") as captured: + verify_slow_result(manifest, expected_commit="a" * 40, expected_run_id=42) + _assert_redacted_exception(captured.value, caplog, sentinel) diff --git a/tools/check_slow_result.py b/tools/check_slow_result.py new file mode 100644 index 0000000..fd00809 --- /dev/null +++ b/tools/check_slow_result.py @@ -0,0 +1,99 @@ +"""Validate slow-integration release evidence.""" + +from __future__ import annotations + +import argparse +import json +import re +from pathlib import Path +from typing import Any + +_COMMIT_SHA = re.compile(r"^[0-9a-f]{40}$") +_PYTHON_RELEASE = re.compile(r"^3\.12\.\d+$") + + +def _required_string(data: dict[str, Any], name: str) -> str: + value = data.get(name) + if not isinstance(value, str) or not value: + raise RuntimeError(f"slow result field {name!r} must be a non-empty string") + return value + + +def _required_positive_integer(data: dict[str, Any], name: str) -> int: + value = data.get(name) + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise RuntimeError(f"slow result field {name!r} must be a positive integer") + return value + + +def verify_slow_result(path: Path, *, expected_commit: str, expected_run_id: int) -> None: + """Require successful full-profile evidence for the exact release commit.""" + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + raise RuntimeError("slow result manifest is unreadable") from None + if not isinstance(data, dict): + raise RuntimeError("slow result manifest must contain a JSON object") + if type(data.get("schema_version")) is not int or data["schema_version"] != 1: + raise RuntimeError("slow result manifest has an unsupported schema version") + + commit = _required_string(data, "commit_sha") + if _COMMIT_SHA.fullmatch(commit) is None or commit != expected_commit: + raise RuntimeError("slow result commit does not match the release commit") + expected_values = { + "profile": "full", + "status": "passed", + "transport": "mysql", + "endpoint_mode": "logical", + "doris_version": "4.0.6", + } + for name, expected in expected_values.items(): + if _required_string(data, name) != expected: + raise RuntimeError(f"slow result field {name!r} must equal {expected!r}") + + if _required_string(data, "ray_version") != "2.55.1": + raise RuntimeError("slow result field 'ray_version' must equal '2.55.1'") + python_version = _required_string(data, "python_version") + if _PYTHON_RELEASE.fullmatch(python_version) is None: + raise RuntimeError("slow result field 'python_version' must be a Python 3.12 release") + if _required_positive_integer(data, "workflow_run_id") != expected_run_id: + raise RuntimeError("slow result workflow run does not match its artifact source") + _required_positive_integer(data, "row_count") + expected_counts = { + "initial_frontend_count": 1, + "initial_backend_count": 3, + "initial_ray_worker_count": 3, + } + for name, expected in expected_counts.items(): + if _required_positive_integer(data, name) != expected: + raise RuntimeError(f"slow result field {name!r} must equal {expected}") + expected_scenarios = [ + "be_failure", + "logical_endpoint_tls", + "mysql_all_workers", + "ray_worker_retry", + ] + scenarios = data.get("scenarios") + if scenarios != expected_scenarios: + raise RuntimeError( + "slow result scenarios do not cover the enterprise-candidate MySQL profile" + ) + + +def main() -> int: + """Validate a manifest supplied by the release workflow.""" + parser = argparse.ArgumentParser() + parser.add_argument("manifest", type=Path) + parser.add_argument("--expected-commit", required=True) + parser.add_argument("--expected-run-id", required=True, type=int) + arguments = parser.parse_args() + verify_slow_result( + arguments.manifest, + expected_commit=arguments.expected_commit, + expected_run_id=arguments.expected_run_id, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())