diff --git a/.github/workflows/code-check.yml b/.github/workflows/code-check.yml index 5ef487bc..c18c4948 100644 --- a/.github/workflows/code-check.yml +++ b/.github/workflows/code-check.yml @@ -8,6 +8,14 @@ on: workflow_dispatch: +# No `concurrency` block with `cancel-in-progress`, deliberately. Cancelling the +# run a push supersedes would halve the clusters two overlapping runs carry, but +# it cannot be made safe: GitHub force-terminates a cancelled job's remaining +# steps after a 5-minute cancellation timeout, `if: always()` included, and an +# S-00 cluster refuses DELETE until it is ACTIVE (~460s). A run cancelled inside +# its first several minutes would die with a cluster it is not yet allowed to +# delete, and nothing on a schedule would come along to reap it. Letting both +# runs finish costs clusters; cancelling them costs stranded clusters. jobs: test-coverage: runs-on: ubuntu-latest @@ -16,6 +24,12 @@ jobs: contents: read actions: write + # One ledger for the whole job, so the cleanup step at the end can reap + # what any of the pytest steps created. See the matching block in + # coverage.yml. + env: + SINGLESTOREDB_TEST_DEPLOYMENT_LOG: ${{ github.workspace }}/deployments.jsonl + services: singlestore: image: ghcr.io/singlestore-labs/singlestoredb-dev:latest @@ -29,12 +43,17 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@v7 with: - fetch-depth: 2 + # Full history, because the change detector below diffs against + # origin/main. A shallow clone does not create that ref -- with + # fetch-depth: 2 every diff died on `fatal: bad revision + # 'origin/main'`, which the detector read as "nothing changed", so + # the management step never ran on a PR. + fetch-depth: 0 - name: Set up Python - uses: actions/setup-python@v4 + uses: actions/setup-python@v7 with: python-version: "3.10" cache: "pip" @@ -78,8 +97,8 @@ jobs: COMMIT_MSG=$(git log -1 --format='%s' HEAD) if [[ "$COMMIT_MSG" =~ ^Prepare\ for\ v[0-9]+\.[0-9]+\.[0-9]+\ release$ ]]; then echo "🚀 Release preparation commit detected: $COMMIT_MSG" - echo "changes-detected=true" >> $GITHUB_OUTPUT - echo "changed-directories=release" >> $GITHUB_OUTPUT + echo "changes-detected=true" >> "$GITHUB_OUTPUT" + echo "changed-directories=release" >> "$GITHUB_OUTPUT" echo "" echo "🎯 RESULT: Full test suite will run for release preparation" exit 0 @@ -94,10 +113,15 @@ jobs: for DIR in $MONITORED_DIRS; do if [ -d "$DIR" ]; then - CHANGED_FILES=$(git diff --name-only $BASE_COMMIT HEAD -- "$DIR" || true) + # No `|| true` here: a git failure means the comparison did not + # happen, and swallowing it silently downgrades the run to the + # no-management path instead of reporting the breakage. + CHANGED_FILES=$(git diff --name-only "$BASE_COMMIT" HEAD -- "$DIR") if [ -n "$CHANGED_FILES" ]; then echo "✅ Changes detected in: $DIR" echo "Files changed:" + # shellcheck disable=SC2001 # prefixing every line, which + # ${var//search/replace} cannot do echo "$CHANGED_FILES" | sed 's/^/ - /' CHANGES_FOUND=true if [ -z "$CHANGED_DIRS" ]; then @@ -115,13 +139,13 @@ jobs: # Set outputs if [ "$CHANGES_FOUND" = true ]; then - echo "changes-detected=true" >> $GITHUB_OUTPUT - echo "changed-directories=$CHANGED_DIRS" >> $GITHUB_OUTPUT + echo "changes-detected=true" >> "$GITHUB_OUTPUT" + echo "changed-directories=$CHANGED_DIRS" >> "$GITHUB_OUTPUT" echo "" echo "🎯 RESULT: Changes detected in monitored directories" else - echo "changes-detected=false" >> $GITHUB_OUTPUT - echo "changed-directories=" >> $GITHUB_OUTPUT + echo "changes-detected=false" >> "$GITHUB_OUTPUT" + echo "changed-directories=" >> "$GITHUB_OUTPUT" echo "" echo "🎯 RESULT: No changes in monitored directories" fi @@ -137,8 +161,12 @@ jobs: - name: Run MySQL protocol tests (with management API) if: steps.check-changes.outputs.changes-detected == 'true' + # -m 'not management_v1' keeps the v2 management coverage while dropping + # the deprecated v1 suite, which coverage.yml runs nightly instead. The + # -m 'not management' steps below need no second term: they already + # exclude everything v1 deploys. run: | - pytest -v --cov=singlestoredb --pyargs singlestoredb.tests + pytest -v -m 'not management_v1' --cov=singlestoredb --pyargs singlestoredb.tests env: COVERAGE_FILE: "coverage-mysql.cov" SINGLESTOREDB_URL: "root:root@127.0.0.1:3307" @@ -171,7 +199,7 @@ jobs: SINGLESTOREDB_FUSION_ENABLE_HIDDEN: "1" - name: Run HTTP protocol tests - # -n 0 overrides the -n 3 in pyproject.toml's addopts: the HTTP/Data API + # -n 0 overrides the -n 2 in pyproject.toml's addopts: the HTTP/Data API # run must be serial. Setup goes over SINGLESTOREDB_INIT_DB_URL (MySQL), # so load_sql takes its `SET GLOBAL HTTP_PROXY_PORT` + `RESTART PROXY` # branch (singlestoredb/tests/utils.py:227) once per worker, and a proxy @@ -195,3 +223,24 @@ jobs: coverage report coverage xml coverage html + + # if: always() is the whole point -- this has to run when the job fails or + # is cancelled, which is what left three clusters billing in run + # 35631802648 (see the matching step in coverage.yml). On a PR the + # management step above only runs when the change detector fires, so most + # runs reach this with an empty ledger and it reports nothing. + # + # always() is not a guarantee, only a best effort: a cancelled job's + # remaining steps are force-terminated after GitHub's 5-minute + # cancellation timeout, so this covers a cancel whose clusters are already + # ACTIVE -- run 35631802648 was cancelled 19 minutes in -- but not one in + # the first several minutes, where DELETE is still being refused when the + # step is killed. That remainder needs `cleanup_deployments.py + # --older-than` run by hand; nothing here is on a schedule. + - name: Terminate any deployment the tests left behind + if: always() + run: | + python -m singlestoredb.tests.cleanup_deployments \ + --ledger "$SINGLESTOREDB_TEST_DEPLOYMENT_LOG" --yes + env: + SINGLESTOREDB_MANAGEMENT_TOKEN: ${{ secrets.CLUSTER_API_KEY }} diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index 6c9546fa..aaf9d757 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -10,6 +10,13 @@ jobs: runs-on: ubuntu-latest environment: Base + # One ledger for the whole job, so the cleanup step below can reap what any + # of the pytest steps created. Per job rather than shared: each job gets its + # own runner and workspace anyway, and keeping the ledgers separate means a + # job's sweep can only ever reach records it wrote itself. + env: + SINGLESTOREDB_TEST_DEPLOYMENT_LOG: ${{ github.workspace }}/deployments.jsonl + services: singlestore: image: ghcr.io/singlestore-labs/singlestoredb-dev:latest @@ -22,10 +29,10 @@ jobs: ROOT_PASSWORD: "root" steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python - uses: actions/setup-python@v4 + uses: actions/setup-python@v7 with: python-version: "3.10" cache: "pip" @@ -36,8 +43,11 @@ jobs: pip install -e ".[dev]" - name: Run MySQL protocol tests + # -m 'not management_v1' keeps the v2 management coverage while dropping + # the deprecated v1 suite; the management-v1-tests job below is where + # that runs. run: | - pytest -v --cov=singlestoredb --pyargs singlestoredb.tests + pytest -v -m 'not management_v1' --cov=singlestoredb --pyargs singlestoredb.tests env: COVERAGE_FILE: "coverage-mysql.cov" SINGLESTOREDB_URL: "root:root@127.0.0.1:3307" @@ -58,7 +68,7 @@ jobs: SINGLESTOREDB_FUSION_ENABLE_HIDDEN: "1" - name: Run HTTP protocol tests - # -n 0 overrides the -n 3 in pyproject.toml's addopts: the HTTP/Data API + # -n 0 overrides the -n 2 in pyproject.toml's addopts: the HTTP/Data API # run must be serial. Setup goes over SINGLESTOREDB_INIT_DB_URL (MySQL), # so load_sql takes its `SET GLOBAL HTTP_PROXY_PORT` + `RESTART PROXY` # branch (singlestoredb/tests/utils.py:227) once per worker, and a proxy @@ -82,3 +92,115 @@ jobs: coverage report coverage xml coverage html + + # if: always() is the whole point -- this has to run when the job is + # cancelled, which is the case that produced the leak. Run 35631802648 + # was cancelled 19 minutes into TestClusterFusion.setUpClass's + # create_cluster(wait_on_active=True, wait_timeout=1200); the log ends at + # '##[error]The operation was canceled.' with no pytest summary and no + # sweep output, so three clusters were left billing with nothing in the + # process having recorded them. The ledger is that record. + # + # Last step in the job so it covers every pytest step above it. Placing + # it after each one instead would add nothing: a cancellation anywhere + # still runs the remaining always() steps. + # + # What always() does not buy is unlimited time. GitHub force-terminates a + # cancelled job's remaining steps after a 5-minute cancellation timeout, + # and an S-00 cluster refuses DELETE until it is ACTIVE (~460s). The leak + # above is covered because it was cancelled 19 minutes in, well past that; + # a cancel in the first several minutes would be killed here still being + # told 400/409, and needs `cleanup_deployments.py --older-than` run by + # hand afterwards. + - name: Terminate any deployment the tests left behind + if: always() + run: | + python -m singlestoredb.tests.cleanup_deployments \ + --ledger "$SINGLESTOREDB_TEST_DEPLOYMENT_LOG" --yes + env: + SINGLESTOREDB_MANAGEMENT_TOKEN: ${{ secrets.CLUSTER_API_KEY }} + + # The deprecated v1 management API. management.version defaults to v2, so this + # is a legacy gate: it runs here nightly rather than on every PR, and it is + # what gets deleted along with management/v1/. Selects both the mocked v1 + # units and the live v1 deployments, plus test_fusion's v1 WORKSPACE grammar. + management-v1-tests: + runs-on: ubuntu-latest + environment: Base + + # Waits for test-coverage rather than running alongside it, so the v1 + # workspace groups are never in flight at the same time as the v2 suite's + # cluster pool -- together they put more on the org than it wants to carry. + # This is a nightly cron, so the extra wall clock costs nothing. + # + # Runs even when test-coverage fails, because this is a legacy gate, not a + # downstream build: a v2 failure above says nothing about the v1 endpoints, + # and skipping v1 for it would hide a v1 regression behind an unrelated one. + # + # !cancelled() rather than always(), which stays true through cancellation + # too. A cancelled run must not go on to start provisioning workspace groups + # here: GitHub force-terminates a cancelled job's remaining steps after a + # 5-minute cancellation timeout, so the ledger sweep below would be killed + # while the new deployments were still pre-ACTIVE and refusing DELETE -- the + # job would strand exactly what it was added to clean up. + needs: test-coverage + if: ${{ !cancelled() }} + + # A ledger of its own, not shared with test-coverage. The jobs no longer + # overlap, but they still run on separate runners with separate workspaces, + # and keeping the ledgers distinct means neither job's sweep can reach the + # other's records. + env: + SINGLESTOREDB_TEST_DEPLOYMENT_LOG: ${{ github.workspace }}/deployments.jsonl + + services: + singlestore: + image: ghcr.io/singlestore-labs/singlestoredb-dev:latest + ports: + - 3307:3306 + - 8081:8080 + - 9081:9081 + env: + SINGLESTORE_LICENSE: ${{ secrets.SINGLESTORE_LICENSE }} + ROOT_PASSWORD: "root" + + steps: + - uses: actions/checkout@v7 + + - name: Set up Python + uses: actions/setup-python@v7 + with: + python-version: "3.10" + cache: "pip" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e ".[dev]" + + - name: Run v1 management API tests + # -n 0 overrides the -n 2 in pyproject.toml's addopts. The parallel + # default is tuned for the v2 management suite's shared cluster pool; the + # v1 classes deploy workspace groups of their own, so two workers here + # put twice that in flight, on top of whatever the v2 job is holding at + # the same time. Serial keeps this job's contribution to the org's + # cluster count to one deployment at a time. + run: | + pytest -v -n 0 -m 'management_v1' --pyargs singlestoredb.tests + env: + SINGLESTOREDB_URL: "root:root@127.0.0.1:3307" + SINGLESTOREDB_PURE_PYTHON: 0 + SINGLESTORE_LICENSE: ${{ secrets.SINGLESTORE_LICENSE }} + SINGLESTOREDB_MANAGEMENT_TOKEN: ${{ secrets.CLUSTER_API_KEY }} + SINGLESTOREDB_FUSION_ENABLE_HIDDEN: "1" + + # See the matching step in test-coverage for why this is if: always(). + # The v1 suite deploys workspace groups, which are the kind that force + # exists for. + - name: Terminate any deployment the tests left behind + if: always() + run: | + python -m singlestoredb.tests.cleanup_deployments \ + --ledger "$SINGLESTOREDB_TEST_DEPLOYMENT_LOG" --yes + env: + SINGLESTOREDB_MANAGEMENT_TOKEN: ${{ secrets.CLUSTER_API_KEY }} diff --git a/.github/workflows/fusion-docs.yml b/.github/workflows/fusion-docs.yml index 75a74ffa..bfbb35d0 100644 --- a/.github/workflows/fusion-docs.yml +++ b/.github/workflows/fusion-docs.yml @@ -14,10 +14,10 @@ jobs: actions: write steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: 3.11 cache: "pip" diff --git a/.github/workflows/pre-commit.yml b/.github/workflows/pre-commit.yml index a9217a93..aa614507 100644 --- a/.github/workflows/pre-commit.yml +++ b/.github/workflows/pre-commit.yml @@ -16,10 +16,10 @@ jobs: - "3.13" steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v7 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v4 + uses: actions/setup-python@v7 with: python-version: ${{ matrix.python-version }} diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index d3a669c1..be006ab2 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -39,7 +39,7 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v7 - name: Install dependencies run: | @@ -49,8 +49,19 @@ jobs: - name: Initialize database id: initialize-database + # A new cluster has exactly one user, admin, which is why that name is + # fixed everywhere below. The project is named here rather than read + # from a repo variable so the deployment target is visible in the + # workflow and does not depend on repository settings. + # + # POST /v2/clusters generates its own admin password and ignores any + # that is sent, so the script resets it to CLUSTER_PASSWORD over SQL + # once the cluster is up. That keeps the credential a secret the runner + # masks everywhere, instead of a job output: the runner refuses to write + # an output whose value is masked, so a generated password could not + # reach these jobs at all. run: | - python resources/create_test_cluster.py --password="${{ secrets.CLUSTER_PASSWORD }}" --token="${{ secrets.CLUSTER_API_KEY }}" --init-sql singlestoredb/tests/test.sql --output=github --expires=2h "python - $GITHUB_WORKFLOW - $GITHUB_RUN_NUMBER" + python resources/create_test_cluster.py --password="${{ secrets.CLUSTER_PASSWORD }}" --token="${{ secrets.CLUSTER_API_KEY }}" --project="Standard Project" --init-sql singlestoredb/tests/test.sql --output=github --expires=2h "python - $GITHUB_WORKFLOW - $GITHUB_RUN_NUMBER" env: PYTHONPATH: ${{ github.workspace }} @@ -73,10 +84,12 @@ jobs: - windows-2022 steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v7 - - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v4 + # This job's matrix varies only over os; cibuildwheel supplies its own + # interpreters, so 3.10 here is just the host Python that drives it. + - name: Set up Python + uses: actions/setup-python@v7 with: python-version: "3.10" cache: "pip" @@ -100,7 +113,7 @@ jobs: - name: Set up QEMU if: runner.os == 'Linux' - uses: docker/setup-qemu-action@v2 + uses: docker/setup-qemu-action@v4 with: platforms: all @@ -118,7 +131,10 @@ jobs: # points --pyargs at the workspace, so that pyproject is the inifile # here. Without the plugin pytest exits on the unknown arguments. CIBW_TEST_REQUIRES: "pytest pytest-xdist" - CIBW_ENVIRONMENT: "SINGLESTOREDB_URL='mysql://${{ secrets.CLUSTER_USER }}:${{ secrets.CLUSTER_PASSWORD }}@${{ needs.setup-database.outputs.cluster-host }}:3306/${{ needs.setup-database.outputs.cluster-database }}?pure_python=0'" + # CLUSTER_PASSWORD has to survive both the userinfo half of the URL + # and the single-quoted shell word cibuildwheel evaluates, so keep the + # secret alphanumeric: no ':', '@', '/', '%' or quote characters. + CIBW_ENVIRONMENT: "SINGLESTOREDB_URL='mysql://admin:${{ secrets.CLUSTER_PASSWORD }}@${{ needs.setup-database.outputs.cluster-host }}:3306/${{ needs.setup-database.outputs.cluster-database }}?pure_python=0'" PYTHONPATH: ${{ github.workspace }} # - name: Build conda @@ -147,14 +163,14 @@ jobs: mv ./wheelhouse/*.whl ./dist/. - name: Archive source dist and wheel - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: name: artifacts-${{ runner.os }} path: dist retention-days: 2 # - name: Archive conda -# uses: actions/upload-artifact@v4 +# uses: actions/upload-artifact@v7 # with: # name: conda-${{ matrix.os }} # path: ./conda-bld @@ -175,22 +191,22 @@ jobs: url: https://pypi.org/p/singlestoredb steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v7 - name: Download Linux wheels and sdist - uses: actions/download-artifact@v4 + uses: actions/download-artifact@v8 with: name: artifacts-Linux path: dist - name: Download Windows wheels and sdist - uses: actions/download-artifact@v4 + uses: actions/download-artifact@v8 with: name: artifacts-Windows path: dist - name: Download Mac wheels and sdist - uses: actions/download-artifact@v4 + uses: actions/download-artifact@v8 with: name: artifacts-macOS path: dist @@ -228,7 +244,7 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v7 - name: Install dependencies run: | @@ -238,14 +254,28 @@ jobs: - name: Drop database if: ${{ always() }} + # The password reaches the script through the environment rather than + # being interpolated into the command, so the shell never sees its + # characters. run: | - python resources/drop_db.py --user "${{ secrets.CLUSTER_USER }}" --password "${{ secrets.CLUSTER_PASSWORD }}" --host "${{ needs.setup-database.outputs.cluster-host }}" --port 3306 --database "${{ needs.setup-database.outputs.cluster-database }}" + python resources/drop_db.py --user admin --password "$CLUSTER_PASSWORD" --host "$CLUSTER_HOST" --port 3306 --database "$CLUSTER_DATABASE" env: PYTHONPATH: ${{ github.workspace }} + CLUSTER_PASSWORD: ${{ secrets.CLUSTER_PASSWORD }} + CLUSTER_HOST: ${{ needs.setup-database.outputs.cluster-host }} + CLUSTER_DATABASE: ${{ needs.setup-database.outputs.cluster-database }} - - name: Shutdown workspace + - name: Shutdown cluster if: ${{ always() }} + # An empty ID would send the DELETE to /v2/clusters/ and leave a live + # cluster behind, so say so loudly instead: at that point the ID has to + # be recovered by hand. --fail-with-body is what makes a refused DELETE + # fail this step rather than printing the error and exiting 0. run: | - curl -H "Accept: application/json" -H "Authorization: Bearer ${{ secrets.CLUSTER_API_KEY }}" -X DELETE "https://api.singlestore.com/v1/workspaces/${{ env.CLUSTER_ID }}" + if [ -z "$CLUSTER_ID" ]; then + echo "::error::No cluster ID from setup-database; the cluster (if any) must be terminated by hand" + exit 1 + fi + curl --fail-with-body -H "Accept: application/json" -H "Authorization: Bearer ${{ secrets.CLUSTER_API_KEY }}" -X DELETE "https://api.singlestore.com/v2/clusters/$CLUSTER_ID?force=true" env: CLUSTER_ID: ${{ needs.setup-database.outputs.cluster-id }} diff --git a/.github/workflows/smoke-test.yml b/.github/workflows/smoke-test.yml index 688a2dc1..7984a806 100644 --- a/.github/workflows/smoke-test.yml +++ b/.github/workflows/smoke-test.yml @@ -12,10 +12,10 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: 3.11 cache: "pip" @@ -28,8 +28,19 @@ jobs: - name: Initialize database id: initialize-database + # A new cluster has exactly one user, admin, which is why that name is + # fixed everywhere below. The project is named here rather than read + # from a repo variable so the deployment target is visible in the + # workflow and does not depend on repository settings. + # + # POST /v2/clusters generates its own admin password and ignores any + # that is sent, so the script resets it to CLUSTER_PASSWORD over SQL + # once the cluster is up. That keeps the credential a secret the runner + # masks everywhere, instead of a job output: the runner refuses to write + # an output whose value is masked, so a generated password could not + # reach these jobs at all. run: | - python resources/create_test_cluster.py --password="${{ secrets.CLUSTER_PASSWORD }}" --token="${{ secrets.CLUSTER_API_KEY }}" --init-sql singlestoredb/tests/test.sql --output=github --expires=2h "python - $GITHUB_WORKFLOW - $GITHUB_RUN_NUMBER" + python resources/create_test_cluster.py --password="${{ secrets.CLUSTER_PASSWORD }}" --token="${{ secrets.CLUSTER_API_KEY }}" --project="Standard Project" --init-sql singlestoredb/tests/test.sql --output=github --expires=2h "python - $GITHUB_WORKFLOW - $GITHUB_RUN_NUMBER" env: PYTHONPATH: ${{ github.workspace }} @@ -48,12 +59,18 @@ jobs: matrix: os: - ubuntu-24.04 + # Every version from the floor in pyproject.toml (requires-python + # >=3.9) up to the newest final release. 3.15 is deliberately absent: + # as of 2026-09-18 it is at rc2 with GA planned for 2026-10-01, and + # setup-python needs allow-prereleases plus an explicit "3.15.0-rc.2" + # to install it at all. Add a bare "3.15" once it ships. python-version: - "3.9" - "3.10" - "3.11" - "3.12" - "3.13" + - "3.14" driver: - mysql - https @@ -100,10 +117,10 @@ jobs: buffered: 1 steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: ${{ matrix.python-version }} cache: "pip" @@ -118,11 +135,14 @@ jobs: run: pytest -v --pyargs singlestoredb.tests.test_basics env: PYTHONPATH: ${{ github.workspace }} - SINGLESTOREDB_URL: "${{ matrix.driver }}://${{ secrets.CLUSTER_USER }}:${{ secrets.CLUSTER_PASSWORD }}@${{ needs.setup-database.outputs.cluster-host }}:3306/${{ needs.setup-database.outputs.cluster-database }}?pure_python=${{ matrix.pure-python }}&buffered=${{ matrix.buffered }}" + # CLUSTER_PASSWORD goes into the userinfo half of a URL here, so it + # has to be free of characters that would need percent-encoding -- + # ':', '@', '/', '%' and the like. Keep the secret alphanumeric. + SINGLESTOREDB_URL: "${{ matrix.driver }}://admin:${{ secrets.CLUSTER_PASSWORD }}@${{ needs.setup-database.outputs.cluster-host }}:3306/${{ needs.setup-database.outputs.cluster-database }}?pure_python=${{ matrix.pure-python }}&buffered=${{ matrix.buffered }}" - name: Run tests if: ${{ matrix.driver == 'https' }} - # -n 0 overrides the -n 3 in pyproject.toml's addopts: the Data API is + # -n 0 overrides the -n 2 in pyproject.toml's addopts: the Data API is # not run in parallel. This job avoids the `RESTART PROXY` hazard the # code-check/coverage HTTP steps hit -- no SINGLESTOREDB_INIT_DB_URL # here, so load_sql's setup connection is itself HTTP and skips that @@ -131,7 +151,7 @@ jobs: run: pytest -v -n 0 --pyargs singlestoredb.tests.test_basics env: PYTHONPATH: ${{ github.workspace }} - SINGLESTOREDB_URL: "${{ matrix.driver }}://${{ secrets.CLUSTER_USER }}:${{ secrets.CLUSTER_PASSWORD }}@${{ needs.setup-database.outputs.cluster-host }}:443/${{ needs.setup-database.outputs.cluster-database }}?pure_python=${{ matrix.pure-python }}&buffered=${{ matrix.buffered }}" + SINGLESTOREDB_URL: "${{ matrix.driver }}://admin:${{ secrets.CLUSTER_PASSWORD }}@${{ needs.setup-database.outputs.cluster-host }}:443/${{ needs.setup-database.outputs.cluster-database }}?pure_python=${{ matrix.pure-python }}&buffered=${{ matrix.buffered }}" shutdown-database: @@ -140,10 +160,10 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: 3.11 cache: "pip" @@ -156,14 +176,28 @@ jobs: - name: Drop database if: ${{ always() }} + # The password reaches the script through the environment rather than + # being interpolated into the command, so the shell never sees its + # characters. run: | - python resources/drop_db.py --user "${{ secrets.CLUSTER_USER }}" --password "${{ secrets.CLUSTER_PASSWORD }}" --host "${{ needs.setup-database.outputs.cluster-host }}" --port 3306 --database "${{ needs.setup-database.outputs.cluster-database }}" + python resources/drop_db.py --user admin --password "$CLUSTER_PASSWORD" --host "$CLUSTER_HOST" --port 3306 --database "$CLUSTER_DATABASE" env: PYTHONPATH: ${{ github.workspace }} + CLUSTER_PASSWORD: ${{ secrets.CLUSTER_PASSWORD }} + CLUSTER_HOST: ${{ needs.setup-database.outputs.cluster-host }} + CLUSTER_DATABASE: ${{ needs.setup-database.outputs.cluster-database }} - - name: Shutdown workspace + - name: Shutdown cluster if: ${{ always() }} + # An empty ID would send the DELETE to /v2/clusters/ and leave a live + # cluster behind, so say so loudly instead: at that point the ID has to + # be recovered by hand. --fail-with-body is what makes a refused DELETE + # fail this step rather than printing the error and exiting 0. run: | - curl -H "Accept: application/json" -H "Authorization: Bearer ${{ secrets.CLUSTER_API_KEY }}" -X DELETE "https://api.singlestore.com/v1/workspaces/${{ env.CLUSTER_ID }}" + if [ -z "$CLUSTER_ID" ]; then + echo "::error::No cluster ID from setup-database; the cluster (if any) must be terminated by hand" + exit 1 + fi + curl --fail-with-body -H "Accept: application/json" -H "Authorization: Bearer ${{ secrets.CLUSTER_API_KEY }}" -X DELETE "https://api.singlestore.com/v2/clusters/$CLUSTER_ID?force=true" env: CLUSTER_ID: ${{ needs.setup-database.outputs.cluster-id }} diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 9d4c6001..b190dd10 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -41,3 +41,7 @@ repos: hooks: - id: mypy additional_dependencies: [types-requests] +- repo: https://github.com/Mateusz-Grzelinski/actionlint-py + rev: v1.7.7.23 + hooks: + - id: actionlint diff --git a/pyproject.toml b/pyproject.toml index 77ff8ca6..25750c42 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -101,13 +101,15 @@ exclude = ["docs*", "resources*", "examples*", "licenses*"] # and honours the xdist_group marks below. It is here rather than in the # invocation because forgetting it costs real money. # -# 3 workers, not `auto`: the ceiling is the management API's tolerance for +# 2 workers, not `auto`: the ceiling is the management API's tolerance for # concurrent provisioning and the org's cluster quota, not this host's CPUs. +# 3 was too many in practice -- the management suite had more clusters in +# flight at once than the org wanted to carry. # # Note that xdist must be installed for pytest to start at all with these set # (`pip install -e ".[test]"`), that SINGLESTOREDB_MANAGEMENT_TRACE's terminal # summary needs -n 0, and that USE_DATA_API=1 in parallel is unverified. -addopts = ["-n", "3", "--dist", "loadgroup"] +addopts = ["-n", "2", "--dist", "loadgroup"] markers = [ "management", "management_v1: exercises the v1 management API, which v2 has replaced. Deselect with -m 'not management_v1'; the v1 endpoints only need a nightly gate now that v2 is the default.", diff --git a/resources/build_docs.py b/resources/build_docs.py index 628f3a1f..5e54b184 100755 --- a/resources/build_docs.py +++ b/resources/build_docs.py @@ -384,6 +384,11 @@ def apply_content_transformations(self, content: str, links: Dict[str, str]) -> # Change workspace.Stage to workspace.stage content = re.sub(r'>workspace\.Stage\.', r'>workspace.stage.', content) + # Change cluster.Stage to cluster.stage. Stage is re-exported from the + # v2 cluster module, so it is documented under both names for as long + # as management/workspace.py is still documented. + content = re.sub(r'>cluster\.Stage\.', r'>cluster.stage.', content) + # Fix class/method links content = re.sub( r'(]+>)?(\s*]*>\s*\s*)([\w\.]+)(\s*\s*)', diff --git a/resources/create_test_cluster.py b/resources/create_test_cluster.py index 186fadfa..28be11c3 100755 --- a/resources/create_test_cluster.py +++ b/resources/create_test_cluster.py @@ -5,10 +5,8 @@ import os import random import re -import secrets import subprocess import sys -import time import uuid from optparse import OptionParser @@ -16,31 +14,39 @@ # Handle command-line options -usage = 'usage: %prog [options] workspace-name' +usage = 'usage: %prog [options] cluster-name' parser = OptionParser(usage=usage) parser.add_option( '-r', '--region', default='AWS::*US East 1*', - help='region pattern or ID', -) -parser.add_option( - '-p', '--password', - default=secrets.token_urlsafe(20) + '-x&$', - help='admin password', + help='region pattern to deploy into, as provider::name ' + '(AWS::*US East 1*); * is a wildcard', ) parser.add_option( '-e', '--expires', default='4h', - help='timestamp when workspace should expire (4h)', + help='when the cluster should expire, as a timestamp or a ' + 'duration such as 4h (4h)', ) parser.add_option( '-s', '--size', default='S-00', - help='size of the workspace (S-00)', + help='size of the cluster (S-00)', +) +parser.add_option( + '-p', '--password', + help='password to give the admin user once the cluster is up; required, ' + 'because the password the API generates cannot be handed to another ' + 'CI job (see below)', ) parser.add_option( '-t', '--token', - help='API key for the workspace management API', + help='API key for the management API', +) +parser.add_option( + '--project', + help='ID or name of the project to deploy into; defaults to the ' + 'organization\'s STANDARD-edition project', ) parser.add_option( '--http-port', type='int', @@ -53,7 +59,7 @@ parser.add_option( '-o', '--output', default='env', choices=['env', 'github', 'json'], - help='report workspace information in the requested format: github, env, json', + help='report cluster information in the requested format: github, env, json', ) parser.add_option( '-d', '--database', @@ -66,120 +72,175 @@ parser.print_help() sys.exit(1) +if not options.password: + print('ERROR: --password is required', file=sys.stderr) + sys.exit(1) + if options.init_sql and not os.path.isfile(options.init_sql): - print('ERROR: Could not locate SQL file: {options.init_sql}', file=sys.stderr) + print(f'ERROR: Could not locate SQL file: {options.init_sql}', file=sys.stderr) sys.exit(1) -# Connect to workspace. This is still the deprecated v1 workspace-group -# grammar because the v1 test suite it sets up needs workspace groups; -# it gets ported to manage_clusters() when that suite goes. Pinned to v1 -# because manage_workspaces() otherwise follows the management.version option. -wm = s2.manage_workspaces(options.token or None, version='v1') +# Pin v2 explicitly rather than following the ambient management.version +# option: this script provisions clusters, which only exist in v2. +mgr = s2.manage_clusters(options.token or None, version='v2') -# Find matching region -if '::' in options.region: - pattern = options.region.replace('*', '.*') - regions = wm.regions - for item in random.sample(regions, k=len(regions)): - region_name = '{}::{}'.format(item.provider, item.name) - if re.match(pattern, region_name): - options.region = item.id - break -if '::' in options.region: +# Find a matching region. A v2 region is identified by the +# (provider, region_name) pair rather than by an ID, so the matched Region +# object is what gets handed to create_cluster. Candidates are shuffled to +# spread deployments across whichever regions match. +# +# The pattern is tried against both the display name and the provider region +# name -- 'US East 1' and 'us-east-1' -- so it does not matter which of the two +# a given listing puts in Region.name. +pattern = options.region.replace('*', '.*') +regions = list(mgr.regions) + + +def candidates(item): + """Return the names ``item`` can be matched by, most specific first.""" + for label in (item.name, item.region_name): + if label: + yield f'{item.provider}::{label}' if '::' in options.region else label + + +region = None +for item in random.sample(regions, k=len(regions)): + if any(re.match(pattern, x) for x in candidates(item)): + region = item + break + +if region is None: print( - 'ERROR: Could not find a region mating the pattern: ' - '{options.region}', file=sys.stderr, + 'ERROR: Could not find a region matching the pattern ' + f'{options.region}; the API reports: ' + + ', '.join(sorted(f'{x.provider}::{x.name}' for x in regions)), + file=sys.stderr, ) sys.exit(1) -# Create workspace group -wg_name = 'Python Client Testing' - -wgs = [x for x in wm.workspace_groups if x.name == wg_name] -if len(wgs) > 1: - print('ERROR: There is more than one workspace group with the specified name.') - sys.exit(1) -elif len(wgs) == 1: - wg = wgs[0] +# Choose a project. projectID is required by POST /v2/clusters and only +# auto-resolves for an organization with a single project, so pick the +# STANDARD-edition one when it was not named explicitly. +if options.project: + project_id = options.project else: - wg = wm.create_workspace_group( - wg_name, - region=options.region, - admin_password=options.password, - # firewall_ranges=requests.get('https://api.github.com/meta').json()['actions'], - firewall_ranges=['0.0.0.0/0'], - allow_all_traffic=True, - ) - -# Make sure the workspace group exists before continuing -timeout = 300 -while timeout > 0 and not [x for x in wm.workspace_groups if x.name == wg_name]: - time.sleep(10) - timeout -= 10 + projects = list(mgr.projects) + standard = [x for x in projects if x.edition == 'STANDARD'] + if not standard: + print( + 'ERROR: No STANDARD-edition project in this organization; pass ' + '--project with one of: ' + + ', '.join(f'{x.name} ({x.id}, {x.edition})' for x in projects), + file=sys.stderr, + ) + sys.exit(1) + project_id = standard[0].id + + +# A cluster name must match [a-z0-9]([a-z0-9-]*[a-z0-9])? and be 1-32 +# characters, so everything outside that alphabet becomes a hyphen, runs of +# hyphens collapse, and the result is truncated with any hyphen the cut +# exposes trimmed off again. +name = re.sub(r'[^a-z0-9]+', '-', args[0].lower()).strip('-')[:32].rstrip('-') +if not name: + print(f'ERROR: Cluster name is empty after cleaning: {args[0]}', file=sys.stderr) + sys.exit(1) -ws_name = re.sub(r'^-|-$', r'', re.sub(r'-+', r'-', re.sub(r'\s+', '-', args[0].lower()))) -ws = wg.create_workspace( - ws_name, +# wait_on_active covers ACTIVE, then the endpoint, then the firewall, so the +# cluster is actually reachable by the time this returns. +cluster = mgr.create_cluster( + name, + region=region, size=options.size, + firewall_ranges=['0.0.0.0/0'], + expires_at=options.expires, + project=project_id, wait_on_active=True, + wait_timeout=1200, ) -# Make sure the endpoint exists before continuing -timeout = 300 -while timeout > 0 and not ws.endpoint: - time.sleep(10) - ws.refresh() - timeout -= 10 - -if not ws.endpoint: - print('ERROR: Endpoint was never activated.') - sys.exit(1) - - -# Extra pause for server to become available -time.sleep(10) - -database = options.database -if not database: - database = 'TEMP_{}'.format(uuid.uuid4()).replace('-', '_') - -host = ws.endpoint +host = cluster.endpoint if ':' in host: host, port = host.split(':', 1) port = int(port) else: port = 3306 -# Print workspace information +database = options.database +if not database: + database = 'TEMP_{}'.format(uuid.uuid4()).replace('-', '_') + +# Report before touching the cluster any further. Everything below can fail +# against a cluster that already exists and is already billing, and the caller's +# only handle on it is the ID reported here -- a CI teardown job with an empty +# cluster-id output would issue its DELETE against /v2/clusters/ and leak the +# cluster it was meant to remove. +# +# No password is reported: the caller passed it in, so it already knows it, and +# under GitHub Actions it is a secret the runner masks on its own. if options.output == 'env': - print(f'CLUSTER_ID={ws.id}') + print(f'CLUSTER_ID={cluster.id}') print(f'CLUSTER_HOST={host}') print(f'CLUSTER_PORT={port}') print(f'CLUSTER_DATABASE={database}') elif options.output == 'github': with open(os.environ['GITHUB_OUTPUT'], 'a') as output: - print(f'cluster-id={ws.id}', file=output) + print(f'cluster-id={cluster.id}', file=output) print(f'cluster-host={host}', file=output) print(f'cluster-port={port}', file=output) print(f'cluster-database={database}', file=output) elif options.output == 'json': print('{') - print(f' "cluster-id": "{ws.id}",') + print(f' "cluster-id": "{cluster.id}",') print(f' "cluster-host": "{host}",') - print(f' "cluster-port": {port}') - print(f' "cluster-database": {database}') + print(f' "cluster-port": {port},') + print(f' "cluster-database": "{database}"') print('}') +# The API generates the admin password and reports it only on the create +# response -- there is no route that will hand it back later, and it is None +# after any refresh(). See item 9 of docs/management-api-audit.md: the API +# accepts an adminPassword on both POST and PATCH and ignores both, which is +# why this is read back rather than set. +generated = cluster.admin_password +if not generated: + print( + 'ERROR: cluster was created without a readable admin password', + file=sys.stderr, + ) + sys.exit(1) + +# Trade the generated password for the caller's, because the generated one +# cannot leave this process. A caller running under GitHub Actions has to mask +# it, and the runner drops any output whose value matches a mask -- "Skip output +# 'cluster-password' since it may contain secret" -- so masking it and passing +# it to another job are mutually exclusive. The password the caller already +# holds has neither problem. +# +# ALTER USER is the statement that works: SET PASSWORD wants a pre-hashed value +# and rejects a literal with '1372: Password hash should be a 41-digit +# hexadecimal number'. Verified against a live S-00 cluster, including that the +# control plane leaves the new password alone afterwards. +password = options.password +escaped = password.replace('\\', '\\\\').replace("'", "\\'") + +with s2.connect( + host=host, port=port, user='admin', + password=generated, connect_timeout=30, +) as conn: + with conn.cursor() as cur: + cur.execute(f"ALTER USER 'admin'@'%' IDENTIFIED BY '{escaped}'") + # Initialize the database if options.init_sql: init_db = [ os.path.join(os.path.dirname(__file__), 'init_db.py'), '--host', str(host), '--port', str(port), - '--user', 'admin', '--password', options.password, + '--user', 'admin', '--password', password, '--database', database, ] diff --git a/resources/drop_test_cluster.py b/resources/drop_test_cluster.py index 16ed7539..6a7105dd 100755 --- a/resources/drop_test_cluster.py +++ b/resources/drop_test_cluster.py @@ -2,7 +2,6 @@ # type: ignore from __future__ import annotations -import re import sys from optparse import OptionParser @@ -10,11 +9,11 @@ # Handle command-line options -usage = 'usage: %prog [options] workspace-id' +usage = 'usage: %prog [options] cluster-id' parser = OptionParser(usage=usage) parser.add_option( '-t', '--token', - help='API key for the workspace management API', + help='API key for the management API', ) (options, args) = parser.parse_args() @@ -23,33 +22,10 @@ sys.exit(1) -# Connect to workspace. This is still the deprecated v1 workspace-group -# grammar because the v1 test suite it sets up needs workspace groups; -# it gets ported to manage_clusters() when that suite goes. Pinned to v1 -# because manage_workspaces() otherwise follows the management.version option. -wm = s2.manage_workspaces(options.token or None, version='v1') +# Pin v2 explicitly rather than following the ambient management.version +# option: clusters only exist in v2. +mgr = s2.manage_clusters(options.token or None, version='v2') -wg_name = 'Python Client Testing' - -wgs = [x for x in wm.workspace_groups if x.name == wg_name] -if len(wgs) > 1: - print('ERROR: There is more than one workspace group with the specified name.') - sys.exit(1) -elif len(wgs) == 0: - print('ERROR: There is no workspace group with the specified name.') - sys.exit(1) -wg = wgs[0] - -ws_name = re.sub(r'^-|-$', r'', re.sub(r'-+', r'-', re.sub(r'\s+', '-', args[0].lower()))) - -wss = [x for x in wg.workspaces if x.name == ws_name] -if len(wss) > 1: - print('ERROR: There is more than one workspace with the specified name.') - sys.exit(1) -elif len(wss) == 0: - print('ERROR: There is no workspace with the specified name.') - sys.exit(1) -ws = wss[0] - -# Terminate workspace -ws.terminate() +# force=True so a cluster with connections still open goes away; this only +# ever runs against clusters this repo's CI created. +mgr.get_cluster(args[0]).terminate(force=True, wait_on_terminated=True) diff --git a/singlestoredb/management/utils.py b/singlestoredb/management/utils.py index bfdcc865..ba6553de 100644 --- a/singlestoredb/management/utils.py +++ b/singlestoredb/management/utils.py @@ -408,6 +408,120 @@ def enable_http_tracing() -> None: requests_log.propagate = True +#: A Go ``time.Time`` rendered by its ``String()`` method: +#: ``2026-09-17 14:42:41.445984 +0000 UTC``. ``GET /v2/clusters/{id}`` reports +#: ``expiresAt`` in this shape while every other timestamp it returns is +#: RFC 3339, and the trailing zone name is not ISO 8601, so the whole value +#: fails to parse and the expiration silently reads as unset. The zone name and +#: the monotonic-clock reading Go appends to some values are both optional. +#: An RFC 3339 ``Z`` counts as an offset here so that shape goes down the same +#: path: its fraction needs the same padding, and until it matched, a value like +#: ``...20.43888Z`` reached the converter with five digits, which only 3.11 and +#: later parse. +_GO_DATETIME_RE = re.compile( + r'^(?P\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}(?:\.\d+)?)' + r'(?:\s*(?P[Zz]|[+-]\d{2}:?\d{2}))?' + r'(?:\s+(?P[A-Za-z]\S*))?' + r'(?:\s+m=\S+)?$', +) + + +def _normalize_datetime(obj: str) -> str: + """ + Return ``obj`` as something :func:`converters.datetime_fromisoformat` reads. + + Handles the two shapes the management API returns -- RFC 3339 and the Go + ``time.Time.String()`` form -- by reducing both to a bare ISO 8601 + timestamp plus an optional numeric offset. Fractional seconds are padded to + microseconds -- Go trims trailing zeros, and ``datetime.fromisoformat`` + accepts only 3 or 6 digits before Python 3.11. + + Parameters + ---------- + obj : str + Timestamp as reported by the API + + Returns + ------- + str + + """ + match = _GO_DATETIME_RE.match(obj.strip()) + if match is None: + # Not a shape this recognizes; hand it over untouched so the converter + # gets its usual chance to make sense of it. + return obj.replace('Z', '') + + stamp = match.group('stamp') + + # Fix datetimes with truncated zeros + if '.' in stamp: + stamp, micros = stamp.split('.', 1) + micros = micros[:6] + '0' * (6 - len(micros)) + stamp = stamp + '.' + micros + + # Go writes the offset without a separator (+0000). Only Python 3.11 and + # later accept that spelling; 3.9 and 3.10 want +00:00, so always emit the + # colon. Z is spelled out for the same reason: nothing before 3.11 reads it. + offset = match.group('offset') or '' + if offset in ('Z', 'z'): + offset = '+00:00' + elif offset and ':' not in offset: + offset = offset[:3] + ':' + offset[3:] + + return stamp + offset + + +def _is_go_zero_time(obj: Union[datetime.date, datetime.datetime]) -> bool: + """ + Return whether ``obj`` is Go's zero time, which means "unset". + + A Go ``time.Time`` that was never assigned renders as January 1 of year 1, + and the API returns that for a field it has no value for -- most visibly an + ``expiresAt`` on a resource that does not expire. It arrives spelled either + way the two timestamp shapes allow: ``0001-01-01T00:00:00Z`` and + ``0001-01-01 00:00:00 +0000 UTC``. Testing the parsed value rather than the + string covers both, along with any offset or monotonic reading that comes + with them. + + Parameters + ---------- + obj : datetime.date or datetime.datetime + Parsed timestamp + + Returns + ------- + bool + + """ + return (obj.year, obj.month, obj.day) == (1, 1, 1) + + +def _as_naive_utc(obj: datetime.datetime) -> datetime.datetime: + """ + Return ``obj`` as a naive UTC datetime. + + A value carrying an offset -- which is every recognized shape, since an + RFC 3339 ``Z`` is normalized to ``+00:00`` -- is shifted onto UTC and + stripped. A value that arrives naive is already meaning UTC and is left + alone. Both end up on the one convention -- otherwise two timestamps read + off the same object could not be compared. + + Parameters + ---------- + obj : datetime.datetime + Parsed timestamp, with or without a timezone + + Returns + ------- + datetime.datetime + + """ + if obj.tzinfo is None: + return obj + return obj.astimezone(datetime.timezone.utc).replace(tzinfo=None) + + def to_datetime( obj: Optional[Union[str, datetime.datetime]], ) -> Optional[datetime.datetime]: @@ -416,20 +530,19 @@ def to_datetime( return None if isinstance(obj, datetime.datetime): return obj - if obj == '0001-01-01T00:00:00Z': - return None - obj = obj.replace('Z', '') - # Fix datetimes with truncated zeros - if '.' in obj: - obj, micros = obj.split('.', 1) - micros = micros + '0' * (6 - len(micros)) - obj = obj + '.' + micros - out = converters.datetime_fromisoformat(obj) + out = converters.datetime_fromisoformat(_normalize_datetime(obj)) if isinstance(out, str): return None + if out is None: + return None + # Before _as_naive_utc: shifting an aware year-1 value onto UTC can carry it + # below datetime.MINYEAR, which raises rather than returning the None this + # value means. + if _is_go_zero_time(out): + return None if isinstance(out, datetime.date) and not isinstance(out, datetime.datetime): return datetime.datetime(out.year, out.month, out.day) - return out + return _as_naive_utc(out) def to_datetime_strict( @@ -440,22 +553,18 @@ def to_datetime_strict( raise TypeError('not possible to convert None to datetime') if isinstance(obj, datetime.datetime): return obj - if obj == '0001-01-01T00:00:00Z': - raise ValueError('not possible to convert 0001-01-01T00:00:00Z to datetime') - obj = obj.replace('Z', '') - # Fix datetimes with truncated zeros - if '.' in obj: - obj, micros = obj.split('.', 1) - micros = micros + '0' * (6 - len(micros)) - obj = obj + '.' + micros - out = converters.datetime_fromisoformat(obj) + out = converters.datetime_fromisoformat(_normalize_datetime(obj)) if not out: raise TypeError('not possible to convert None to datetime') if isinstance(out, str): raise ValueError('value cannot be str') + # See to_datetime: checked here rather than after the UTC shift, which can + # raise on a year-1 value. + if _is_go_zero_time(out): + raise ValueError(f'not possible to convert {obj} to datetime') if isinstance(out, datetime.date) and not isinstance(out, datetime.datetime): return datetime.datetime(out.year, out.month, out.day) - return out + return _as_naive_utc(out) def from_datetime( diff --git a/singlestoredb/management/v1/workspace.py b/singlestoredb/management/v1/workspace.py index 718b292f..de5938da 100644 --- a/singlestoredb/management/v1/workspace.py +++ b/singlestoredb/management/v1/workspace.py @@ -850,7 +850,15 @@ def terminate( raise ManagementError( msg='No workspace manager is associated with this object.', ) - self._manager._delete(f'workspaceGroups/{self.id}', params=dict(force=force)) + # 'true'/'false', not the bool: requests renders a bool param with + # str(), so force=True went out as force=True. Workspace.terminate + # above already builds the lowercase form by hand; this matches it. + # force is what makes a group with live workspaces in it go away, so + # the value being read is not optional. + self._manager._delete( + f'workspaceGroups/{self.id}', + params=dict(force='true' if force else 'false'), + ) if wait_on_terminated: remaining = float(wait_timeout) while True: diff --git a/singlestoredb/management/v2/cluster.py b/singlestoredb/management/v2/cluster.py index 511b7cac..df51765f 100644 --- a/singlestoredb/management/v2/cluster.py +++ b/singlestoredb/management/v2/cluster.py @@ -611,7 +611,18 @@ def update( allow_all_traffic : bool, optional Allow all traffic to the cluster admin_password : str, optional - Admin password for the cluster + Admin password for the cluster. + + .. warning:: This is ignored, exactly as it is on + ``POST /v2/clusters``. ``PATCH /v2/clusters/{id}`` accepts the + field and does not honor it: a live probe found the patched value + refused with ``1045: Access denied`` while the password the + original create generated kept working. So the admin password + cannot be set after the fact either -- the only value that + authenticates is the generated one + :attr:`Cluster.admin_password` carried on the create response. + See item 9 of ``docs/management-api-audit.md``. The field is + still sent in case the API starts honoring it. expires_at : str, optional Timestamp of when the cluster will expire. Expiration time can be specified as a timestamp or a duration. @@ -704,7 +715,12 @@ def terminate( """ manager = self._require_manager() - manager._delete(f'clusters/{self.id}', params=dict(force=force)) + # 'true'/'false', not the bool: requests renders a bool param with + # str(), so force=True went out as force=True. + manager._delete( + f'clusters/{self.id}', + params=dict(force='true' if force else 'false'), + ) if wait_on_terminated: remaining = float(wait_timeout) while True: diff --git a/singlestoredb/tests/cleanup_deployments.py b/singlestoredb/tests/cleanup_deployments.py index 7b8d745c..6b715431 100644 --- a/singlestoredb/tests/cleanup_deployments.py +++ b/singlestoredb/tests/cleanup_deployments.py @@ -42,20 +42,31 @@ are tracked as they are created and swept per test class by ``conftest.py``, which cannot see -- or touch -- another run's deployments. -Why strays keep appearing: that tracking, the per-class sweep and this script -all live on the ``versioned-management-api`` branch and nowhere else. A run -from ``main`` has only ``tearDownClass``, so a killed run or a ``setUpClass`` -that raises leaks a workspace group permanently, and ``main`` still uses names -this script only knows through :data:`LEGACY_PATTERNS`. Until the sweep is on -the default branch, expect to run this by hand. +The exception, and the reason this is wired into CI, is ``--ledger``. A run +with ``SINGLESTOREDB_TEST_DEPLOYMENT_LOG`` set records every creation to a +JSONL file as it happens (``utils.ledger_pending``/``ledger_live``/ +``ledger_gone``), so a run that was killed outright leaves an exact list of +what it made:: + + python -m singlestoredb.tests.cleanup_deployments --ledger deployments.jsonl + +That mode replaces *both* guards above -- the name patterns and the age +filter. Neither is needed, because the ledger names the deployments rather +than guessing at them, and neither is safe: a ledger entry is minutes old by +construction, so the age filter would spare everything it lists. What keeps +such a run off other people's deployments is that it only ever touches ids and +names the ledger records, and that each CI job writes its own ledger. """ import argparse import datetime +import json +import os import re import sys import warnings from collections.abc import Container from typing import Any +from typing import Dict from typing import List from typing import Optional from typing import Tuple @@ -78,12 +89,32 @@ #: anything younger could belong to a run in progress. DEFAULT_MIN_AGE_HOURS = 6.0 +#: How long to keep retrying a deployment the API will not delete yet. This is +#: the end of the line -- nothing runs after this tool -- so it does not borrow +#: ``utils.TERMINATE_RETRY_TIMEOUT``, which is deliberately short so the sweep +#: between test classes cannot stall the suite. Here a deployment may still be +#: coming up, ``DELETE`` is refused until it is, and an S-00 cluster reaching +#: ACTIVE is ~460s at worst, so anything shorter than a full provision leaves it +#: billing. An upper bound on retrying, not a promise of it: the whole budget is +#: available when the job that calls this ends normally or fails, but a +#: *cancelled* job's steps are force-terminated after GitHub's 5-minute +#: cancellation timeout, so a cancel early in a provision gets killed here +#: regardless of what this says. The only cost of the larger budget is the CI +#: step's wall clock. +TERMINATE_TIMEOUT = 600.0 + #: Names the suite generates. Anchored, because these run against a real #: organization: a pattern that matched a name someone chose by hand would #: terminate a deployment that is not ours. PATTERNS = [ # test_management_v1.py / test_management_v2.py fixtures re.compile(r'^(wg|ws|cl)-test-[A-Za-z0-9_-]+$'), + # TestWorkspace.test_update renames its live group from wg-test- to + # wg-foo- and never renames it back, so the group carries this name + # for the rest of the class. No pattern matched it, which made a group + # stranded after that test invisible to this sweep -- it would pile up + # while the tool reported nothing. + re.compile(r'^wg-foo-[A-Za-z0-9_-]+$'), re.compile(r'^starter-(ws|cl)-test-[A-Za-z0-9_-]+$'), # test_fusion.py fixtures re.compile(r'^[A-C] Fusion Testing [0-9a-f]+$'), @@ -254,7 +285,7 @@ def keep(obj: Any) -> bool: if 'cluster' in kinds or 'starter-cluster' in kinds: try: - clusters = s2.manage_clusters(version='v2') + clusters = _manager('v2') except Exception as exc: print(f'! Could not reach management API v2: {exc}', file=sys.stderr) else: @@ -274,14 +305,7 @@ def keep(obj: Any) -> bool: if 'workspace-group' in kinds or 'starter-workspace' in kinds: try: - # v1 is deprecated, and asking for it here is the point: workspace - # groups exist nowhere else, so the warning is noise on every run. - with warnings.catch_warnings(): - warnings.filterwarnings( - 'ignore', category=DeprecationWarning, - message='.*manage_workspaces.*', - ) - workspaces = s2.manage_workspaces(version='v1') + workspaces = _manager('v1') except Exception as exc: print(f'! Could not reach management API v1: {exc}', file=sys.stderr) else: @@ -304,12 +328,297 @@ def keep(obj: Any) -> bool: return found, spared, unmatched +# +# Ledger mode +# +# What this exists for: GH Actions run 35631802648, job ``test-coverage``, was +# cancelled 19 minutes into a ``create_cluster(wait_on_active=True, +# wait_timeout=1200)`` and the log ends at ``##[error]The operation was +# canceled.`` with no pytest summary and no sweep output at all. Three clusters +# were live and no in-process handler ever ran. Reading a file written as the +# clusters were created is the only way to know that from another process. +# + +#: How each ledger kind is resolved back to a live object: the management API +#: version that owns it, the point lookup for a record that has an id, and the +#: listing to search by name for a ``pending`` record that never got one. +#: +#: The kinds are the values of ``utils._KIND_BY_CLASS``; a kind this does not +#: know is reported rather than skipped, since the alternative is silently not +#: reaping it. +LEDGER_KINDS = { + 'cluster': ( + 'v2', 'get_cluster', lambda mgr: mgr.clusters, + ), + 'starter_cluster': ( + 'v2', 'get_starter_cluster', lambda mgr: mgr.starter_clusters, + ), + 'workspace_group': ( + 'v1', 'get_workspace_group', lambda mgr: mgr.workspace_groups, + ), + 'workspace': ( + # WorkspaceManager has no `workspaces` of its own, so the search goes + # group by group -- the same walk utils._CREATORS uses. + 'v1', 'get_workspace', + lambda mgr: [w for g in mgr.workspace_groups for w in g.workspaces], + ), + 'starter_workspace': ( + 'v1', 'get_starter_workspace', lambda mgr: mgr.starter_workspaces, + ), +} + + +def _manager(version: str) -> Any: + """Management API manager for ``'v1'`` or ``'v2'``.""" + if version == 'v2': + return s2.manage_clusters(version='v2') + # v1 is deprecated, and asking for it here is the point: workspace groups + # exist nowhere else, so the warning is noise on every run. + with warnings.catch_warnings(): + warnings.filterwarnings( + 'ignore', category=DeprecationWarning, + message='.*manage_workspaces.*', + ) + return s2.manage_workspaces(version='v1') + + +def fold_ledger(lines: Any) -> List[Dict[str, Any]]: + """ + Reduce ledger records to the deployments that should still be live. + + The ledger is append-only and written from several processes (one per xdist + worker), so it is a history, not a state: a deployment shows up as + ``pending``, then ``live`` once it has an id, then ``gone`` once something + terminated it. Folding keeps whatever the last event for a deployment was + not ``gone``. + + A ``pending`` is keyed by ``(kind, name)`` because that is all it has; the + matching ``live`` retires it and re-keys on the id. So the two records a + normal creation writes collapse to one entry, and a ``pending`` left + standing means the creator was interrupted before it returned -- the + cancelled-mid-``wait_on_active`` case, resolvable only by name. + + Order is creation order, since dicts preserve insertion order and a + deployment's key is first inserted when it first appears. The caller + reverses it, so a workspace goes before the group that holds it, matching + ``utils.cleanup_tracked()``. + + Malformed lines are skipped with a warning rather than aborting: this runs + as the last step of a CI job, and one truncated line -- a process killed + between the ``write`` and the ``fsync``, which the per-line fsync makes + unlikely but not impossible -- must not stop the rest from being reaped. + """ + live: Dict[Any, Dict[str, Any]] = {} + + for lineno, line in enumerate(lines, start=1): + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + except ValueError as exc: + print( + f'! ledger line {lineno} is not JSON, skipping it: {exc}', + file=sys.stderr, + ) + continue + if not isinstance(record, dict): + continue + + event = record.get('event') + kind = record.get('kind') + name = record.get('name') + ident = record.get('id') + + by_name = ('name', kind, name) + by_id = ('id', kind, ident) + + if event == 'pending': + if name is not None: + live.setdefault(by_name, record) + elif event == 'live': + live.pop(by_name, None) + if ident is not None: + live[by_id] = record + elif name is not None: + # No id in the record: keep it findable by name rather than + # dropping it. Should not happen, but losing the deployment is + # the expensive direction. + live[by_name] = record + elif event == 'gone': + if ident is not None: + live.pop(by_id, None) + live.pop(by_name, None) + + return list(live.values()) + + +def read_ledger(path: str) -> List[Dict[str, Any]]: + """ + Fold the ledger at ``path``, newest first. + + A missing file is not an error: the variable can be set on a job whose + tests created nothing, and a CI cleanup step that failed in that case would + turn every such run red. + """ + if not os.path.exists(path): + print(f'No ledger at {path}; nothing this run created was recorded.') + return [] + with open(path, encoding='utf-8') as file: + records = fold_ledger(file) + # Newest first, so a workspace is terminated before its group. + records.reverse() + return records + + +def find_ledger_leftovers( + path: str, +) -> Tuple[List[Tuple[str, Any]], List[str], List[str]]: + """ + Resolve the ledger's still-live records to live deployment objects. + + Returns + ------- + (List[Tuple[str, Any]], List[str], List[str]) + The deployments to terminate, labels for the records that resolved to + nothing -- already gone, so nothing to do -- and labels for the ones + that could not be resolved *and* could still be live, which is what + makes the run exit non-zero. + + A 404 from the point lookup means the deployment is already gone, which is + the common case: the ledger records every creation, and a run that finished + normally terminated all of them. Anything else -- a transport failure, an + unknown kind -- goes in the third list, because "could not tell" and "not + there" must not read the same when the difference is a cluster billing. + """ + from singlestoredb.exceptions import ManagementError + + found: List[Tuple[str, Any]] = [] + gone: List[str] = [] + unresolved: List[str] = [] + + managers: Dict[str, Any] = {} + + def manager_for(version: str) -> Any: + if version not in managers: + managers[version] = _manager(version) + return managers[version] + + for record in read_ledger(path): + kind = record.get('kind') + name = record.get('name') + ident = record.get('id') + label = '{} {} ({})'.format( + str(kind).replace('_', ' '), name or '', ident or 'no id', + ) + + if kind not in LEDGER_KINDS: + unresolved.append(f'{label}: unknown kind {kind!r}') + continue + version, lookup_name, listing = LEDGER_KINDS[kind] + + try: + mgr = manager_for(version) + except Exception as exc: + unresolved.append( + f'{label}: could not reach management API ' + f'{version}: {exc}', + ) + continue + + obj = None + try: + if ident is not None: + obj = getattr(mgr, lookup_name)(ident) + else: + # A `pending` record: the creator never returned an id, so the + # only handle on it is the name. Matched over the listing + # exactly as utils._recover_orphan does. + for candidate in listing(mgr): + if getattr(candidate, 'name', None) == name: + obj = candidate + break + except ManagementError as exc: + if exc.errno == 404: + gone.append(label) + continue + unresolved.append(f'{label}: {exc}') + continue + except Exception as exc: + unresolved.append(f'{label}: {exc}') + continue + + if obj is None: + gone.append(label) + elif getattr(obj, 'terminated_at', None) is not None: + gone.append(f'{label} (already terminated)') + else: + found.append((label, obj)) + + return found, gone, unresolved + + +def _run_ledger_sweep(path: str, yes: bool) -> int: + """Report, and with ``yes`` terminate, everything the ledger still lists.""" + leftovers, gone, unresolved = find_ledger_leftovers(path) + + print( + f'Ledger {path}: {len(leftovers)} still live, {len(gone)} already ' + f'gone, {len(unresolved)} unresolved.\n', + ) + + if unresolved: + print( + f'{len(unresolved)} ledger record(s) could not be resolved, so ' + 'they may still be live:', + ) + for label in unresolved: + print(f' ? {label}') + print() + + if not leftovers: + # Non-zero only for the records whose state is unknown: a clean run + # whose sweep already terminated everything must not fail the job. + print('Nothing left behind by this run.') + return 1 if unresolved else 0 + + print(f'{len(leftovers)} deployment(s) left behind by this run:') + for label, _ in leftovers: + print(f' - {label}') + + if not yes: + print('\nDry run; pass --yes to terminate these.') + return 0 + + from singlestoredb.tests import utils + + failed = 0 + for label, obj in leftovers: + try: + utils.terminate(obj, timeout=TERMINATE_TIMEOUT) + except Exception as exc: + failed += 1 + print(f'✗ {label}: {exc}') + else: + print(f'✓ terminated {label}') + + return 1 if (failed or unresolved) else 0 + + def main(argv: Optional[List[str]] = None) -> int: parser = argparse.ArgumentParser(description=__doc__.split('\n\n')[1]) parser.add_argument( '--yes', action='store_true', help='actually terminate; without this the run only reports', ) + parser.add_argument( + '--ledger', metavar='PATH', + help='sweep exactly what the run that wrote this JSONL ledger created ' + '(see SINGLESTOREDB_TEST_DEPLOYMENT_LOG). Replaces both the name ' + 'patterns and the age filter, which a ledger makes unnecessary ' + 'and which would in any case spare everything in it for being ' + 'minutes old. This is the mode CI runs as an if: always() step', + ) parser.add_argument( '--older-than', type=float, default=DEFAULT_MIN_AGE_HOURS, metavar='HOURS', @@ -357,6 +666,21 @@ def main(argv: Optional[List[str]] = None) -> int: ) args = parser.parse_args(argv) + # --ledger is a different question entirely -- "what did *this* run make?" + # rather than "what looks stranded?" -- so it does not compose with the + # name and age guards, and saying so beats silently ignoring them. + if args.ledger: + for flag, value in ( + ('--older-than', args.older_than != DEFAULT_MIN_AGE_HOURS), + ('--since', args.since is not None), + ('--any-name', args.any_name), + ('--kind', bool(args.kinds)), + ('--show-unmatched', args.show_unmatched), + ): + if value: + parser.error(f'{flag} does not apply with --ledger') + return _run_ledger_sweep(args.ledger, args.yes) + kinds = args.kinds or list(KINDS) leftovers, spared, unmatched = find_leftovers( @@ -415,7 +739,10 @@ def main(argv: Optional[List[str]] = None) -> int: failed = 0 for label, obj in leftovers: try: - utils.terminate(obj) + # Same budget as the ledger sweep: --since or --older-than 0 can + # select a deployment that is still provisioning, and nothing runs + # after this either. + utils.terminate(obj, timeout=TERMINATE_TIMEOUT) except Exception as exc: failed += 1 print(f'✗ {label}: {exc}') diff --git a/singlestoredb/tests/conftest.py b/singlestoredb/tests/conftest.py index 9d426a64..c1e3d819 100644 --- a/singlestoredb/tests/conftest.py +++ b/singlestoredb/tests/conftest.py @@ -297,6 +297,75 @@ def on_sigterm(signum: int, frame: Any) -> None: logger.debug('Not the main thread; no SIGTERM sweep installed') +#: Key the workers stash their stranded deployment labels under in +#: ``config.workeroutput``. +_STRANDED_KEY = 'singlestoredb_stranded_deployments' + + +def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: + """ + Sweep in an xdist worker, and hand what survived to the controller. + + ``addopts`` is ``-n 2`` (``pyproject.toml``), so under the default the + sweep and its ``STILL LIVE`` banner run in a worker, whose stdout the + controller discards. A leak was therefore silent even when the sweep did + run and fail -- the one case the banner exists to make loud. + + ``config.workeroutput`` is the channel xdist provides for exactly this, and + it only exists in a worker: its absence is the ``-n 0`` case, where + ``pytest_unconfigure`` prints directly to a terminal someone is reading and + nothing here is needed. + + Sweeping here rather than leaving it all to ``pytest_unconfigure`` is what + makes the labels available at all. xdist's own + ``pytest_sessionfinish`` is a hookwrapper that sends ``workeroutput`` after + yielding, so anything written to it from this hook is still included -- + but ``pytest_unconfigure`` runs after the send, so a sweep that waited + until then would have nothing left to report. The sweep is idempotent (a + successful one empties ``_tracked``), so the later call simply finds + nothing to do. + """ + workeroutput = getattr(session.config, 'workeroutput', None) + if workeroutput is None: + return + + _sweep_live_deployments() + + try: + workeroutput[_STRANDED_KEY] = _test_utils().tracked_labels() + except Exception: # pragma: no cover - shutdown path + pass + + +def pytest_testnodedown(node: Any, error: Any) -> None: + """ + Report, on the controller, what a worker could not terminate. + + Runs in the controller process, whose output the user actually sees. The + worker's own banner went to a captured stream; this is the copy that gets + read. + """ + stranded = getattr(node, 'workeroutput', {}).get(_STRANDED_KEY) or [] + if not stranded: + return + + print('\n' + '!' * 70) + print( + f'STILL LIVE on {node.gateway.id} -- these deployments could not be ' + 'terminated and are costing money:', + ) + for label in stranded: + print(f' - {label}') + print( + 'Reap them with: python -m singlestoredb.tests.cleanup_deployments ' + '--yes', + ) + print('!' * 70) + logger.error( + f'{len(stranded)} deployment(s) left live by {node.gateway.id}', + ) + + def pytest_unconfigure(config: pytest.Config) -> None: """ Pytest hook that runs after all tests complete. diff --git a/singlestoredb/tests/test_fusion.py b/singlestoredb/tests/test_fusion.py index 248259dc..f130e467 100644 --- a/singlestoredb/tests/test_fusion.py +++ b/singlestoredb/tests/test_fusion.py @@ -969,31 +969,36 @@ def setUpClass(cls): # US-only: no test here asserts anything about these groups' regions, # and creation in some non-US regions fails with a control-plane 500. us_regions = [x for x in mgr.regions if x.name.startswith('US')] - wg = mgr.create_workspace_group( - f'A Fusion Testing {cls.id}', - region=random.choice(us_regions), - firewall_ranges=[], - ) - cls.workspace_groups.append(wg) - wg = mgr.create_workspace_group( - f'B Fusion Testing {cls.id}', - region=random.choice(us_regions), - firewall_ranges=[], - ) - cls.workspace_groups.append(wg) - wg = mgr.create_workspace_group( - f'C Fusion Testing {cls.id}', - region=random.choice(us_regions), - firewall_ranges=[], - ) - cls.workspace_groups.append(wg) + for letter in ('A', 'B', 'C'): + cls.workspace_groups.append( + mgr.create_workspace_group( + f'{letter} Fusion Testing {cls.id}', + region=random.choice(us_regions), + firewall_ranges=[], + expires_at=utils.DEPLOYMENT_EXPIRES_AT, + ), + ) @classmethod def tearDownClass(cls): - if not cls.dbexisted: - utils.drop_database(cls.dbname) + # Deployments first, and each one guarded. Dropping the database first + # -- as this used to -- meant a database error aborted the teardown + # before a single group was terminated, and an unguarded loop meant a + # failure on the first group abandoned the other two. Three workspace + # groups is the most expensive thing this file leaks. while cls.workspace_groups: - cls.workspace_groups.pop().terminate(force=True) + group = cls.workspace_groups.pop() + try: + group.terminate(force=True) + except Exception: + # Left to utils.cleanup_tracked, which retries and then reports + # it; raising here would replace the test's own failure. + pass + try: + if not cls.dbexisted: + utils.drop_database(cls.dbname) + except Exception: + pass def setUp(self): self.enabled = os.environ.get('SINGLESTOREDB_FUSION_ENABLED') @@ -1411,6 +1416,12 @@ class _ClusterFusionMixin: cluster-less ones start immediately and no class deploys more than it reads. + Only :class:`TestClusterFusionSuspendResume` still names a prefix, because + it is the only one left that both needs a cluster up front and mutates it. + :class:`TestClusterFusion` reads without mutating and so borrows from + ``utils.shared_clusters``; the lifecycle suites create their own clusters + in the test bodies, those creates being the subject under test. + Not a ``TestCase``, and named with a leading underscore: pytest collects any ``Test``-prefixed ``TestCase`` subclass it can reach, so a base that was either would run every inherited test a second time under a fixture @@ -1481,6 +1492,7 @@ def setUpClass(cls): f'{prefix}-fusion-cluster-{cls.id}', region=region, size='S-00', + expires_at=utils.DEPLOYMENT_EXPIRES_AT, project=cls.project_id, wait_on_active=True, wait_timeout=1200, @@ -1489,8 +1501,9 @@ def setUpClass(cls): @classmethod def tearDownClass(cls): - if not cls.dbexisted: - utils.drop_database(cls.dbname) + # Clusters before the database: a drop_database failure used to abort + # the teardown before anything was terminated, leaving three clusters + # to the sweep. while cls.clusters: cluster = cls.clusters.pop() try: @@ -1503,6 +1516,11 @@ def tearDownClass(cls): cluster.terminate(force=True) except Exception: pass + try: + if not cls.dbexisted: + utils.drop_database(cls.dbname) + except Exception: + pass def setUp(self): self.enabled = os.environ.get('SINGLESTOREDB_FUSION_ENABLED') @@ -1530,24 +1548,46 @@ def tearDown(self): @pytest.mark.management +@pytest.mark.xdist_group(utils.SHARED_CLUSTER_STAGE_GROUP) class TestClusterFusion(_ClusterFusionMixin, unittest.TestCase): """ - ``SHOW CLUSTERS`` against three deployed clusters. - - Three of them so the ``LIKE``/``ORDER BY``/``LIMIT`` assertions have - something to sort. Nothing here mutates a cluster, which is what makes the - fixture shareable -- ``SUSPEND``/``RESUME`` cannot share it and deploys its - own in :class:`TestClusterFusionSuspendResume`. + ``SHOW CLUSTERS`` against the shared cluster pool. + + Borrows rather than deploying: nothing here mutates a cluster -- these are + four ``SHOW`` statements -- which is the condition ``utils.shared_clusters`` + asks of a consumer. ``SUSPEND``/``RESUME`` cannot borrow and deploys its own + in :class:`TestClusterFusionSuspendResume`. + + Three of them, which is one more than the pool was built for, so the + ``LIKE``/``ORDER BY``/``LIMIT`` assertions have something to sort. Joining + the Stage group rather than the Jobs one because Stage already asks for two: + the pool grows to the largest request, so this costs that group one extra + cluster instead of three, and the Jobs group is left at one. + + Every assertion here is scoped to ``utils.shared_cluster_pattern()`` and + counted against ``utils.shared_cluster_names()``, never a literal. The pool + is shared and grows to whatever the largest request in the process turns out + to be, so a hardcoded 3 would break the day a class asks for four -- and + would break silently, as a row count, which is the failure this class had + before when its count depended on other classes' clusters leaving the list + endpoint in time. """ - fixture_prefixes = ('a', 'b', 'c') + #: Borrowed, so kept out of ``clusters``, which ``tearDownClass`` + #: terminates. A pool cluster must outlive the class that used it. + pool_clusters: List[Any] = [] + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.pool_clusters = utils.shared_clusters(3) def test_show_clusters(self): self.cur.execute('show clusters') names = [x[0] for x in self.cur.fetchall()] assert self.cur.description[0][0] == 'Name' - for prefix in ('a', 'b', 'c'): - assert f'{prefix}-fusion-cluster-{self.id}' in names, names + for cluster in type(self).pool_clusters: + assert cluster.name in names, names def test_show_clusters_columns(self): self.cur.execute('show clusters') @@ -1562,42 +1602,45 @@ def test_show_clusters_columns(self): 'TerminatedAt', ], cols + cluster = type(self).pool_clusters[0] rows = {x[0]: x for x in self.cur.fetchall()} - row = rows[f'a-fusion-cluster-{self.id}'] + row = rows[cluster.name] # Region is the provider slug; Cluster has no region object at v2. assert row[2], row assert row[5], row # ProjectName, not the ID: the column reports the name the project - # listing gives for the ID the cluster was deployed into. - project = type(self).manager.projects[type(self).project_id] - assert row[9] == project.name, row + # listing gives for the ID the cluster was deployed into. Read back + # from the cluster rather than from this class's own project_id -- + # the pool resolves its project independently, and asserting against + # the borrower's copy would be asserting the two resolutions agree. + expected = type(self).manager.get_cluster(cluster.id).project + assert row[9] == expected.name, row def test_show_clusters_like(self): - self.cur.execute(f'show clusters like "a-fusion-cluster-{self.id}"') + one = type(self).pool_clusters[0].name + self.cur.execute(f'show clusters like "{one}"') names = [x[0] for x in self.cur.fetchall()] - assert names == [f'a-fusion-cluster-{self.id}'], names + assert names == [one], names - self.cur.execute(f'show clusters like "%-fusion-cluster-{self.id}"') + self.cur.execute( + f'show clusters like "{utils.shared_cluster_pattern()}"', + ) names = [x[0] for x in self.cur.fetchall()] - assert len(names) == 3, names + assert sorted(names) == sorted(utils.shared_cluster_names()), names def test_show_clusters_order_by_and_limit(self): - self.cur.execute( - f'show clusters like "%-fusion-cluster-{self.id}" order by name', - ) + pattern = utils.shared_cluster_pattern() + + self.cur.execute(f'show clusters like "{pattern}" order by name') names = [x[0] for x in self.cur.fetchall()] assert names == sorted(names), names - self.cur.execute( - f'show clusters like "%-fusion-cluster-{self.id}" ' - 'order by name desc', - ) + self.cur.execute(f'show clusters like "{pattern}" order by name desc') names = [x[0] for x in self.cur.fetchall()] assert names == sorted(names, reverse=True), names self.cur.execute( - f'show clusters like "%-fusion-cluster-{self.id}" ' - 'order by name limit 2', + f'show clusters like "{pattern}" order by name limit 2', ) names = [x[0] for x in self.cur.fetchall()] assert len(names) == 2, names @@ -1833,16 +1876,36 @@ def test_create_cluster_without_project(self): 'this test is for', ) - with self.assertRaises(Exception): - self.cur.execute( - f'create cluster "{name}" in region "{region.region_name}"', - ) + live = [] + try: + with self.assertRaises(Exception): + self.cur.execute( + f'create cluster "{name}" in region ' + f'"{region.region_name}"', + ) + finally: + # One listing, serving both purposes: the assertion that nothing + # was created, and the cleanup for when something was. The test + # only passes if this comes back empty, so the terminate below + # fires exactly when the assertion is about to fail -- which is + # also the only case where a cluster exists. + # + # Belt and braces rather than the only cleanup, contrary to what + # this used to claim: the handler reaches the API through + # ClusterManager.create_cluster (fusion/handlers/cluster.py), which + # is the method utils._CREATORS wraps, so a cluster created here is + # tracked and ledgered like any other and the sweep would find it. + # Terminating it now just means not waiting for the sweep. + live = [ + x for x in mgr.clusters + if x.name == name and x.terminated_at is None + ] + for cluster in live: + try: + utils.terminate(cluster) + except Exception: + pass - # Nothing should have been created - live = [ - x for x in mgr.clusters - if x.name == name and x.terminated_at is None - ] assert not live, live def test_create_cluster_named_project(self): @@ -1880,18 +1943,23 @@ def test_create_cluster_named_project(self): assert mgr.get_cluster(cluster_id).project.id == project.id finally: - # force=True: the cluster is still PENDING, having never been - # waited out, and a termination request is refused otherwise. + # utils.terminate, not a bare terminate(force=True). The cluster is + # PENDING, never having been waited out, and force does not make a + # pre-ACTIVE deployment deletable -- the API refuses it with a 400 + # or a 409 either way (see utils.terminate, which retries exactly + # that). A single forced DELETE here was therefore the likeliest + # outcome, swallowed by the except, leaving the cluster to the + # sweep; utils.terminate retries until it lands. if cluster_id is not None: try: - mgr.get_cluster(cluster_id).terminate(force=True) + utils.terminate(mgr.get_cluster(cluster_id)) except Exception: pass else: for cluster in mgr.clusters: if cluster.name == name and cluster.terminated_at is None: try: - cluster.terminate(force=True) + utils.terminate(cluster) except Exception: pass diff --git a/singlestoredb/tests/test_management_utils.py b/singlestoredb/tests/test_management_utils.py index b7b25f1e..b1e784a8 100644 --- a/singlestoredb/tests/test_management_utils.py +++ b/singlestoredb/tests/test_management_utils.py @@ -8,8 +8,10 @@ only because that is where the bugs were found. """ import datetime +import json import os import pathlib +import shutil import tempfile import unittest from types import SimpleNamespace @@ -17,7 +19,10 @@ from unittest.mock import patch from singlestoredb.exceptions import ManagementError +from singlestoredb.management.utils import _normalize_datetime from singlestoredb.management.utils import normalize_remote_path +from singlestoredb.management.utils import to_datetime +from singlestoredb.management.utils import to_datetime_strict from singlestoredb.tests.utils import counting_file_space from singlestoredb.tests.utils import counting_stage @@ -949,8 +954,18 @@ def _restore(self): self.utils._in_flight.clear() self.utils._in_flight.extend(self.saved_in_flight) - def _deployment(self, name, terminated_at=None, state='ACTIVE'): - """A stand-in that is not a Mock, so tracking does not skip it.""" + def _deployment( + self, name, terminated_at=None, state='ACTIVE', classname=None, + ): + """ + A stand-in that is not a Mock, so tracking does not skip it. + + ``classname`` renames the class, which is how the ledger decides a + kind (``utils._KIND_BY_CLASS`` is keyed by class name). The default + ``Deployment`` is deliberately *not* a ledger kind, so the tests that + only care about tracking write no ledger records even when one is + configured. + """ class Deployment: def __init__(self): self.name = name @@ -966,6 +981,8 @@ def refresh(self): def terminate(self, force=False): self.terminated_with = force + if classname: + Deployment.__name__ = classname return Deployment() def test_mocked_deployments_are_not_tracked(self): @@ -1071,7 +1088,7 @@ def create_then_fail_waiting(recv, name, **kwargs): raise ManagementError(msg=f'Exceeded waiting time for {name}') wrapped = self.utils._tracking_wrapper( - create_then_fail_waiting, lambda recv: recv.clusters, + create_then_fail_waiting, 'cluster', lambda recv: recv.clusters, ) with self.assertRaises(ManagementError): wrapped(receiver, 'cl-test-shared-0-abc', wait_on_active=True) @@ -1096,7 +1113,7 @@ def interrupted(recv, name, **kwargs): raise KeyboardInterrupt wrapped = self.utils._tracking_wrapper( - interrupted, lambda recv: recv.clusters, + interrupted, 'cluster', lambda recv: recv.clusters, ) with self.assertRaises(KeyboardInterrupt): wrapped(receiver, 'cl-1') @@ -1118,7 +1135,7 @@ def test_a_mocked_receiver_does_not_track_what_it_returns(self): for value in (returned, 'sentinel'): wrapped = self.utils._tracking_wrapper( lambda recv, name, value=value, **kwargs: value, - lambda recv: [], + 'cluster', lambda recv: [], ) self.assertIs(wrapped(mgr, 'my-cluster'), value) @@ -1133,7 +1150,7 @@ def test_a_real_receiver_still_tracks_what_it_returns(self): returned._manager = None wrapped = self.utils._tracking_wrapper( - lambda recv, name, **kwargs: returned, lambda recv: [], + lambda recv, name, **kwargs: returned, 'cluster', lambda recv: [], ) wrapped(mgr, 'cl-1') self.assertEqual(self.utils.tracked_labels(), ["Deployment 'cl-1'"]) @@ -1144,7 +1161,9 @@ def test_a_mocked_receiver_is_not_searched_for_orphans(self): def boom(recv, name, **kwargs): raise ManagementError(msg='boom') - wrapped = self.utils._tracking_wrapper(boom, lambda recv: recv.clusters) + wrapped = self.utils._tracking_wrapper( + boom, 'cluster', lambda recv: recv.clusters, + ) with self.assertRaises(ManagementError): wrapped(MagicMock(), 'cl-1') self.assertEqual(self.utils._tracked, []) @@ -1162,7 +1181,7 @@ def boom(recv, name, **kwargs): def finder(recv): raise AssertionError('recovery called the live API') - wrapped = self.utils._tracking_wrapper(boom, finder) + wrapped = self.utils._tracking_wrapper(boom, 'cluster', finder) with self.assertRaises(ManagementError): wrapped(receiver, 'cl-1') self.assertEqual(self.utils._tracked, []) @@ -1184,7 +1203,7 @@ def create_then_wait(recv, name, **kwargs): raise AssertionError('the process would have been killed here') wrapped = self.utils._tracking_wrapper( - create_then_wait, lambda recv: recv.clusters, + create_then_wait, 'cluster', lambda recv: recv.clusters, ) with self.assertRaises(AssertionError): wrapped(receiver, 'cl-1', wait_on_active=True) @@ -1205,7 +1224,7 @@ def finder(recv): made = self._deployment('cl-1') wrapped = self.utils._tracking_wrapper( - lambda recv, name, **kwargs: made, finder, + lambda recv, name, **kwargs: made, 'cluster', finder, ) self.assertIs(wrapped(receiver, 'cl-1'), made) self.assertEqual(self.utils._in_flight, []) @@ -1216,7 +1235,9 @@ def boom(recv, name, **kwargs): receiver.clusters = [self._deployment('cl-2')] with self.assertRaises(ManagementError): - self.utils._tracking_wrapper(boom, finder)(receiver, 'cl-2') + self.utils._tracking_wrapper( + boom, 'cluster', finder, + )(receiver, 'cl-2') self.assertEqual(self.utils._in_flight, []) # The orphan was recovered once, not once per code path. self.assertEqual(len(self.utils._tracked), 2) @@ -1226,7 +1247,7 @@ def create(recv, name, **kwargs): raise AssertionError(str(self.utils._in_flight)) wrapped = self.utils._tracking_wrapper( - create, lambda recv: recv.clusters, + create, 'cluster', lambda recv: recv.clusters, ) with self.assertRaises(AssertionError) as raised: wrapped(MagicMock(), 'cl-1') @@ -1296,7 +1317,7 @@ def test_every_creation_method_is_wrapped(self): import importlib self.utils.install_deployment_tracking() - for module_name, class_name, method_name, _ in self.utils._CREATORS: + for module_name, class_name, method_name, _, _ in self.utils._CREATORS: klass = getattr(importlib.import_module(module_name), class_name) method = getattr(klass, method_name, None) self.assertIsNotNone( @@ -1313,7 +1334,9 @@ def test_every_creator_takes_name_first_and_has_a_finder(self): import importlib import inspect - for module_name, class_name, method_name, finder in \ + from singlestoredb.tests import cleanup_deployments + + for module_name, class_name, method_name, kind, finder in \ self.utils._CREATORS: klass = getattr(importlib.import_module(module_name), class_name) method = getattr(klass, method_name) @@ -1328,6 +1351,590 @@ def test_every_creator_takes_name_first_and_has_a_finder(self): 'a failed create would not be recoverable', ) self.assertTrue(callable(finder)) + # The ledger's `pending` record carries this kind, and the reaper + # resolves it through cleanup_deployments.LEDGER_KINDS. A kind + # neither side knows would make a cancelled create unreapable, + # which is the whole point of the ledger. + self.assertIn( + kind, set(self.utils._KIND_BY_CLASS.values()), + f'{class_name}.{method_name} has an unknown ledger kind', + ) + self.assertIn(kind, cleanup_deployments.LEDGER_KINDS) + + +class TestDeploymentLedger(TestDeploymentTracking): + """ + The on-disk ledger that makes a killed run's deployments reapable. + + Inherits ``TestDeploymentTracking``'s fixtures for the module globals and + the non-Mock deployment stand-in. It re-runs that class's tests with a + ledger configured, which is worth having: those tests all use the default + ``Deployment`` classname, so they also pin that a ledger being configured + changes nothing about the in-memory behaviour. + """ + + def setUp(self): + super().setUp() + self.dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.dir, True) + self.ledger = os.path.join(self.dir, 'deployments.jsonl') + patcher = patch.dict( + os.environ, {self.utils.LEDGER_ENV_VAR: self.ledger}, + ) + patcher.start() + self.addCleanup(patcher.stop) + + def records(self): + """Every record in the ledger, in the order it was written.""" + if not os.path.exists(self.ledger): + return [] + with open(self.ledger) as file: + return [json.loads(x) for x in file if x.strip()] + + def events(self): + return [(x['event'], x.get('kind'), x.get('name')) for x in + self.records()] + + # + # Writing + # + + def test_no_ledger_is_written_without_the_environment_variable(self): + """Opt-in is the whole contract: a local run must behave exactly as it + did before, with no file appearing anywhere.""" + with patch.dict(os.environ, {}, clear=False): + del os.environ[self.utils.LEDGER_ENV_VAR] + self.utils.track(self._deployment('cl-1', classname='Cluster')) + self.assertFalse(os.path.exists(self.ledger)) + + def test_a_mocked_creation_writes_nothing(self): + """The unit tests drive the creators with patched transports. Recording + those would have the reaper chasing ids that never existed, and -- worse + -- exit non-zero on every one it could not resolve.""" + wrapped = self.utils._tracking_wrapper( + lambda recv, name, **kwargs: MagicMock(), + 'cluster', lambda recv: [], + ) + wrapped(MagicMock(), 'cl-1') + self.utils.track(MagicMock()) + self.assertEqual(self.records(), []) + + def test_a_real_creation_writes_pending_then_live(self): + """In that order, and with the pending written before the creator is + even called: the window this closes is the one where the POST has landed + and nothing in the process knows an id yet.""" + made = self._deployment('cl-1', classname='Cluster') + seen = [] + + def create(recv, name, **kwargs): + # What the ledger holds *during* the wait, which is where the + # cancelled job died. + seen.extend(self.events()) + return made + + receiver = SimpleNamespace( + _get=object(), _post=object(), _delete=object(), + ) + wrapped = self.utils._tracking_wrapper( + create, 'cluster', lambda recv: [], + ) + wrapped(receiver, 'cl-1') + + self.assertEqual(seen, [('pending', 'cluster', 'cl-1')]) + self.assertEqual( + self.events(), [ + ('pending', 'cluster', 'cl-1'), + ('live', 'cluster', 'cl-1'), + ], + ) + self.assertEqual(self.records()[1]['id'], 'cl-1') + + def test_the_pending_name_comes_from_the_keyword_too(self): + receiver = SimpleNamespace( + _get=object(), _post=object(), _delete=object(), + ) + self.utils._tracking_wrapper( + lambda recv, name, **kwargs: None, 'workspace_group', + lambda recv: [], + )(receiver, name='wg-1') + self.assertEqual( + self.events(), [('pending', 'workspace_group', 'wg-1')], + ) + + def test_a_create_that_dies_mid_wait_leaves_pending_with_no_gone(self): + """The reported failure, as the ledger sees it. The creator raises and + the orphan is not in the listing yet, so nothing else is ever written -- + and that lone `pending` is what the reaper resolves by name.""" + receiver = SimpleNamespace( + _get=object(), _post=object(), _delete=object(), clusters=[], + ) + + def create_then_fail_waiting(recv, name, **kwargs): + raise ManagementError(msg=f'Exceeded waiting time for {name}') + + wrapped = self.utils._tracking_wrapper( + create_then_fail_waiting, 'cluster', lambda recv: recv.clusters, + ) + with self.assertRaises(ManagementError): + wrapped(receiver, 'a-fusion-cluster-1f2e', wait_on_active=True) + + self.assertEqual( + self.events(), + [('pending', 'cluster', 'a-fusion-cluster-1f2e')], + ) + + def test_a_recovered_orphan_is_recorded_live(self): + """``_recover_orphan`` goes through ``track()``, so the id it digs out + of the listing reaches the ledger and the reaper can use the point + lookup instead of searching by name.""" + orphan = self._deployment('cl-1', classname='Cluster') + receiver = SimpleNamespace( + _get=object(), _post=object(), _delete=object(), + clusters=[orphan], + ) + + def boom(recv, name, **kwargs): + raise ManagementError(msg='boom') + + with self.assertRaises(ManagementError): + self.utils._tracking_wrapper( + boom, 'cluster', lambda recv: recv.clusters, + )(receiver, 'cl-1') + + self.assertEqual( + self.events(), [ + ('pending', 'cluster', 'cl-1'), + ('live', 'cluster', 'cl-1'), + ], + ) + + def test_a_successful_sweep_appends_gone(self): + obj = self._deployment('cl-1', classname='Cluster') + self.utils.track(obj) + self.assertEqual(len(self.utils.cleanup_tracked()), 1) + self.assertEqual( + self.events(), [ + ('live', 'cluster', 'cl-1'), + ('gone', 'cluster', 'cl-1'), + ], + ) + + def test_a_deployment_already_gone_is_recorded_gone(self): + """A test that terminated in its own teardown: the sweep finds it gone + rather than terminating it, and the record still has to be closed or + the reaper spends a lookup on it and reports it unresolved.""" + obj = self._deployment( + 'cl-1', terminated_at='now', classname='Cluster', + ) + self.utils.track(obj) + self.assertEqual(self.utils.cleanup_tracked(), []) + self.assertEqual( + [x['event'] for x in self.records()], ['live', 'gone'], + ) + + def test_a_failed_terminate_writes_no_gone(self): + """The deployment is still live and still billing, so the reaper must + still see it.""" + obj = self._deployment('cl-1', classname='Cluster') + + def boom(force=False): + raise ManagementError(errno=500, msg='boom') + + obj.terminate = boom + self.utils.track(obj) + self.assertEqual(self.utils.cleanup_tracked(), []) + self.assertEqual([x['event'] for x in self.records()], ['live']) + + def test_untrack_records_gone_only_for_something_tracked(self): + obj = self._deployment('cl-1', classname='Cluster') + self.utils.untrack(obj) + self.assertEqual(self.records(), []) + + self.utils.track(obj) + self.utils.untrack(obj) + self.assertEqual([x['event'] for x in self.records()], ['live', 'gone']) + + def test_a_write_failure_is_logged_and_not_raised(self): + """This sits on the creation path of every management test: an + unwritable ledger must cost a warning, not a failed test run.""" + with patch.dict( + os.environ, + {self.utils.LEDGER_ENV_VAR: os.path.join(self.dir, 'no', 'such')}, + ): + with self.assertLogs(self.utils.logger, 'WARNING') as logs: + self.utils.track(self._deployment('cl-1', classname='Cluster')) + self.assertIn('deployment ledger', logs.output[0]) + + # + # Folding + # + + def fold(self, *lines): + from singlestoredb.tests import cleanup_deployments + return cleanup_deployments.fold_ledger(lines) + + def test_folding_keeps_only_what_is_not_gone(self): + kept = self.fold( + json.dumps(dict(event='pending', kind='cluster', name='cl-1')), + json.dumps( + dict(event='live', kind='cluster', name='cl-1', id='id-1'), + ), + json.dumps( + dict(event='gone', kind='cluster', name='cl-1', id='id-1'), + ), + # Created and never terminated. + json.dumps(dict(event='pending', kind='cluster', name='cl-2')), + json.dumps( + dict(event='live', kind='cluster', name='cl-2', id='id-2'), + ), + # Interrupted before it returned: pending only. + json.dumps(dict(event='pending', kind='cluster', name='cl-3')), + ) + self.assertEqual( + [(x['event'], x.get('id'), x['name']) for x in kept], + [('live', 'id-2', 'cl-2'), ('pending', None, 'cl-3')], + ) + + def test_a_live_record_retires_its_pending(self): + """Otherwise the reaper resolves the same cluster twice -- once by id + and once by name -- and reports two.""" + kept = self.fold( + json.dumps(dict(event='pending', kind='cluster', name='cl-1')), + json.dumps( + dict(event='live', kind='cluster', name='cl-1', id='id-1'), + ), + ) + self.assertEqual(len(kept), 1) + self.assertEqual(kept[0]['id'], 'id-1') + + def test_gone_cancels_a_pending_that_never_went_live(self): + kept = self.fold( + json.dumps(dict(event='pending', kind='cluster', name='cl-1')), + json.dumps(dict(event='gone', kind='cluster', name='cl-1')), + ) + self.assertEqual(kept, []) + + def test_the_same_name_in_two_kinds_is_two_deployments(self): + """`cl-test-abc` as a cluster and as a workspace are different things, + and a `gone` for one must not clear the other.""" + kept = self.fold( + json.dumps(dict(event='pending', kind='cluster', name='x')), + json.dumps(dict(event='pending', kind='workspace', name='x')), + json.dumps(dict(event='gone', kind='cluster', name='x')), + ) + self.assertEqual([x['kind'] for x in kept], ['workspace']) + + def test_a_malformed_line_is_skipped_rather_than_fatal(self): + """A truncated last line -- a process killed between the write and the + fsync -- must not cost the reaper every other record.""" + kept = self.fold( + json.dumps(dict(event='pending', kind='cluster', name='cl-1')), + '{"event": "pending", "kin', + '', + '[]', + ) + self.assertEqual([x['name'] for x in kept], ['cl-1']) + + def test_reading_reverses_into_newest_first(self): + """A workspace has to be terminated before the group that holds it, the + same ordering ``cleanup_tracked`` uses.""" + from singlestoredb.tests import cleanup_deployments + with open(self.ledger, 'w') as file: + for kind, name in ( + ('workspace_group', 'wg-1'), ('workspace', 'ws-1'), + ): + file.write( + json.dumps(dict(event='pending', kind=kind, name=name)) + + '\n', + ) + self.assertEqual( + [x['name'] for x in cleanup_deployments.read_ledger(self.ledger)], + ['ws-1', 'wg-1'], + ) + + # + # Resolving, with the management API stubbed out + # + + def stub_managers(self, **attrs): + """Patch the reaper's manager lookup with a namespace.""" + from singlestoredb.tests import cleanup_deployments + mgr = SimpleNamespace(**attrs) + patcher = patch.object( + cleanup_deployments, '_manager', lambda version: mgr, + ) + patcher.start() + self.addCleanup(patcher.stop) + return cleanup_deployments, mgr + + def write_ledger(self, *records): + with open(self.ledger, 'w') as file: + for record in records: + file.write(json.dumps(record) + '\n') + + def test_an_id_that_404s_is_treated_as_already_gone(self): + """The common case by far: the ledger records every creation, and a run + that ended normally terminated all of them. A clean sweep must exit 0 + and terminate nothing.""" + def get_cluster(ident): + raise ManagementError(errno=404, msg='not found') + + mod, _ = self.stub_managers(get_cluster=get_cluster) + self.write_ledger( + dict(event='live', kind='cluster', name='cl-1', id='id-1'), + ) + found, gone, unresolved = mod.find_ledger_leftovers(self.ledger) + self.assertEqual(found, []) + self.assertEqual(len(gone), 1) + self.assertEqual(unresolved, []) + self.assertEqual(mod.main(['--ledger', self.ledger, '--yes']), 0) + + def test_a_live_id_is_resolved_and_terminated(self): + obj = self._deployment('cl-1', classname='Cluster') + mod, _ = self.stub_managers(get_cluster=lambda ident: obj) + self.write_ledger( + dict(event='live', kind='cluster', name='cl-1', id='id-1'), + ) + self.assertEqual(mod.main(['--ledger', self.ledger, '--yes']), 0) + self.assertTrue(obj.terminated_with) + + def test_a_dry_run_terminates_nothing(self): + obj = self._deployment('cl-1', classname='Cluster') + mod, _ = self.stub_managers(get_cluster=lambda ident: obj) + self.write_ledger( + dict(event='live', kind='cluster', name='cl-1', id='id-1'), + ) + self.assertEqual(mod.main(['--ledger', self.ledger]), 0) + self.assertIsNone(obj.terminated_with) + + def test_a_pending_record_is_resolved_by_name_over_the_listing(self): + """No id was ever returned, so the listing is the only handle -- the + same match ``_recover_orphan`` makes, and the case a cancelled + ``wait_on_active`` leaves.""" + wanted = self._deployment('a-fusion-cluster-1f2e', classname='Cluster') + other = self._deployment('someone-elses', classname='Cluster') + mod, _ = self.stub_managers(clusters=[other, wanted]) + self.write_ledger( + dict(event='pending', kind='cluster', name='a-fusion-cluster-1f2e'), + ) + self.assertEqual(mod.main(['--ledger', self.ledger, '--yes']), 0) + self.assertTrue(wanted.terminated_with) + self.assertIsNone(other.terminated_with) + + def test_a_pending_name_absent_from_the_listing_is_gone(self): + """The POST never landed, so there is nothing to reap and nothing to + complain about.""" + mod, _ = self.stub_managers(clusters=[]) + self.write_ledger(dict(event='pending', kind='cluster', name='cl-1')) + found, gone, unresolved = mod.find_ledger_leftovers(self.ledger) + self.assertEqual((found, len(gone), unresolved), ([], 1, [])) + + def test_an_already_terminated_deployment_is_not_terminated_again(self): + obj = self._deployment( + 'cl-1', terminated_at='now', classname='Cluster', + ) + mod, _ = self.stub_managers(get_cluster=lambda ident: obj) + self.write_ledger( + dict(event='live', kind='cluster', name='cl-1', id='id-1'), + ) + self.assertEqual(mod.main(['--ledger', self.ledger, '--yes']), 0) + self.assertIsNone(obj.terminated_with) + + def test_a_lookup_failure_that_is_not_a_404_exits_non_zero(self): + """"Could not tell" and "not there" must not read the same when the + difference is a cluster billing.""" + def get_cluster(ident): + raise ManagementError(errno=500, msg='gateway sulked') + + mod, _ = self.stub_managers(get_cluster=get_cluster) + self.write_ledger( + dict(event='live', kind='cluster', name='cl-1', id='id-1'), + ) + found, gone, unresolved = mod.find_ledger_leftovers(self.ledger) + self.assertEqual((found, gone), ([], [])) + self.assertEqual(len(unresolved), 1) + self.assertEqual(mod.main(['--ledger', self.ledger, '--yes']), 1) + + def test_an_unknown_kind_is_reported_rather_than_skipped(self): + mod, _ = self.stub_managers() + self.write_ledger(dict(event='live', kind='mystery', name='x', id='1')) + _, _, unresolved = mod.find_ledger_leftovers(self.ledger) + self.assertEqual(len(unresolved), 1) + self.assertIn('unknown kind', unresolved[0]) + + def test_a_missing_ledger_is_not_an_error(self): + """The variable is set for a whole job, including steps whose tests + create nothing. Failing there would turn those runs red.""" + from singlestoredb.tests import cleanup_deployments + missing = os.path.join(self.dir, 'never-written.jsonl') + self.assertEqual(cleanup_deployments.read_ledger(missing), []) + self.assertEqual( + cleanup_deployments.main(['--ledger', missing, '--yes']), 0, + ) + + def test_the_sweep_waits_out_a_provision_rather_than_the_class_budget(self): + """The whole point of the ledger is a job cancelled inside + ``wait_on_active``, whose cluster is minutes from deletable. Borrowing + ``utils.TERMINATE_RETRY_TIMEOUT`` -- short so the per-class sweep cannot + stall the suite -- would exhaust the budget and leave it billing, and + nothing runs after this to try again.""" + from singlestoredb.tests import utils + obj = self._deployment('cl-1', classname='Cluster') + mod, _ = self.stub_managers(get_cluster=lambda ident: obj) + self.write_ledger( + dict(event='live', kind='cluster', name='cl-1', id='id-1'), + ) + + calls = [] + patcher = patch.object( + utils, 'terminate', + lambda obj, **kwargs: calls.append(kwargs), + ) + patcher.start() + self.addCleanup(patcher.stop) + + self.assertEqual(mod.main(['--ledger', self.ledger, '--yes']), 0) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0]['timeout'], mod.TERMINATE_TIMEOUT) + self.assertGreater(calls[0]['timeout'], utils.TERMINATE_RETRY_TIMEOUT) + + def test_ledger_mode_refuses_the_guards_it_replaces(self): + """Silently ignoring --older-than would read as a safety guard that is + not there.""" + from singlestoredb.tests import cleanup_deployments + for extra in ( + ['--older-than', '0'], ['--any-name'], + ['--kind', 'cluster'], ['--show-unmatched'], + ): + with self.assertRaises(SystemExit): + cleanup_deployments.main( + ['--ledger', self.ledger] + extra, + ) + + +class TestTerminateRetry(unittest.TestCase): + """ + ``utils.terminate()``'s bounded retry for a deployment the API will not + delete yet. + + A deployment killed mid-provision is PENDING/TRANSITIONING and the DELETE + comes back 400 or 409. Nothing retried that: Manager.RETRY_STATUSES is + {429, 500, 502, 503, 504}, so the per-class sweep warned, the session-end + sweep tried once more and the cluster stayed up. + """ + + def setUp(self): + from singlestoredb.tests import utils + self.utils = utils + self.slept = [] + # A fake clock, not just a stubbed sleep: the retry budget is measured + # with time.monotonic(), so a sleep that does not advance it makes the + # deadline unreachable and the loop only ends when the stub runs out of + # refusals. That is the opposite of what the budget test asserts. + self.now = 0.0 + + def sleep(seconds): + self.slept.append(seconds) + self.now += seconds + + for name, value in ( + ('sleep', sleep), ('monotonic', lambda: self.now), + ): + patcher = patch(f'time.{name}', value) + patcher.start() + self.addCleanup(patcher.stop) + + def _refuser(self, *errnos): + """A deployment whose terminate raises these in turn, then succeeds.""" + class Deployment: + attempts = 0 + terminated_with = None + + def terminate(inner, force=False): + inner.attempts += 1 + if inner.attempts <= len(errnos): + raise ManagementError( + errno=errnos[inner.attempts - 1], + msg='still provisioning', + ) + inner.terminated_with = force + + return Deployment() + + def test_a_400_is_retried_until_it_succeeds(self): + obj = self._refuser(400, 409) + self.utils.terminate(obj) + self.assertEqual(obj.attempts, 3) + self.assertTrue(obj.terminated_with) + self.assertEqual(self.slept, [15.0, 15.0]) + + def test_a_404_is_not_retried(self): + """It is already gone; retrying would burn the whole budget waiting for + something that is not coming back.""" + obj = self._refuser(404) + with self.assertRaises(ManagementError): + self.utils.terminate(obj) + self.assertEqual(obj.attempts, 1) + self.assertEqual(self.slept, []) + + def test_a_5xx_is_not_retried_here(self): + """The transport already retried it; another round trip from this layer + is not what fixes it.""" + obj = self._refuser(503) + with self.assertRaises(ManagementError): + self.utils.terminate(obj) + self.assertEqual(obj.attempts, 1) + + def test_the_budget_is_bounded_and_the_error_is_re_raised(self): + """Raising is what keeps the deployment in ``_tracked``, so the + end-of-session sweep gets another go at it.""" + obj = self._refuser(*([409] * 100)) + with self.assertRaises(ManagementError): + self.utils.terminate(obj, timeout=45.0, interval=15.0) + self.assertEqual(obj.attempts, 3) + self.assertEqual(self.slept, [15.0, 15.0]) + + def test_a_starter_kind_is_terminated_without_force(self): + """StarterWorkspace.terminate / StarterCluster.terminate take no + arguments at all.""" + class Starter: + called = False + + def terminate(inner): + inner.called = True + + obj = Starter() + self.utils.terminate(obj) + self.assertTrue(obj.called) + + def test_force_is_passed_when_the_signature_accepts_it(self): + """``force`` is what makes a workspace group with live workspaces in it + go away, so this is not cosmetic.""" + seen = [] + + class Group: + def terminate(inner, force=False): + seen.append(force) + + self.utils.terminate(Group()) + self.assertEqual(seen, [True]) + + def test_a_type_error_from_inside_terminate_is_not_a_second_delete(self): + """The signature is inspected rather than discovered by catching + TypeError from the call. The old ``except TypeError`` also caught one + raised *inside* a terminate that did accept force, and retried without + it -- two DELETEs, the second unforced, which is exactly the shape that + leaves a workspace group behind.""" + calls = [] + + class Group: + def terminate(inner, force=False): + calls.append(force) + raise TypeError('something inside went wrong') + + with self.assertRaises(TypeError): + self.utils.terminate(Group()) + self.assertEqual(calls, [True]) class TestSharedClusterPool(unittest.TestCase): @@ -1340,6 +1947,21 @@ def setUp(self): from singlestoredb.tests import utils self.utils = utils + # Redirected before anything can create a cluster: the stand-in + # manager's create_cluster calls the real utils.track, which ledgers, + # and _pool_id is the live one, so under CI these mocked units used to + # append `id-of-cl-test-shared-N-` to the job's real + # ledger. The cleanup step then could not resolve those ids and exited + # non-zero on every run, burying any genuine unresolved record. + tmp = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, tmp, True) + patcher = patch.dict( + os.environ, + {utils.LEDGER_ENV_VAR: os.path.join(tmp, 'deployments.jsonl')}, + ) + patcher.start() + self.addCleanup(patcher.stop) + self.saved_pool = list(utils._pool) self.saved_skip = utils._pool_skip self.saved_tracked = list(utils._tracked) @@ -1504,6 +2126,53 @@ def test_an_explicit_project_does_not_need_a_standard_one(self): self.assertEqual(self.created[0][2]['project'], 'chosen-project') + def test_pool_clusters_are_given_an_expiry(self): + # The only cleanup that survives the process being killed, so it has to + # be on the POST rather than left to the sweep. + with self._patched(): + self.utils.shared_clusters(2) + + self.assertEqual( + [x[2].get('expires_at') for x in self.created], + [self.utils.DEPLOYMENT_EXPIRES_AT] * 2, + ) + + def test_the_pattern_matches_the_pool_and_is_scoped_to_this_process(self): + with self._patched(): + self.utils.shared_clusters(2) + + pattern = self.utils.shared_cluster_pattern() + prefix, _, suffix = pattern.partition('%') + + # A LIKE pattern, so assert it the way the server would read it: + # every pool name matches, and the suffix is the per-process id that + # keeps another run's pool from matching. + for name in self.utils.shared_cluster_names(): + self.assertTrue(name.startswith(prefix), (name, pattern)) + self.assertTrue(name.endswith(suffix), (name, pattern)) + + self.assertEqual(suffix, f'-{self.utils._pool_id}') + + # And another process's pool does not: same prefix, different id. + other = f'cl-test-shared-0-{"f" * 8}' + self.assertTrue(other.startswith(prefix), (other, pattern)) + self.assertFalse(other.endswith(suffix), (other, pattern)) + + def test_the_names_follow_the_pool_as_it_grows(self): + # Read at assertion time rather than cached, so a class that asks for + # more clusters later cannot leave an exact-count expectation stale. + with self._patched(): + self.utils.shared_clusters(1) + self.assertEqual(len(self.utils.shared_cluster_names()), 1) + + self.utils.shared_clusters(3) + self.assertEqual(len(self.utils.shared_cluster_names()), 3) + + self.assertEqual( + self.utils.shared_cluster_names(), + [x[0] for x in self.created], + ) + class TestClearStage(unittest.TestCase): """ @@ -1676,6 +2345,14 @@ def test_the_default_spares_anything_a_run_could_still_own(self): # Not zero: a default that swept every match would make running this # during a test run destructive. self.assertGreaterEqual(self.mod.DEFAULT_MIN_AGE_HOURS, 1) + # Nothing runs after this tool, so its terminate budget has to cover a + # full provision (~460s for an S-00 cluster reaching ACTIVE) rather than + # the per-class budget, which is short on purpose. + from singlestoredb.tests import utils + self.assertGreater( + self.mod.TERMINATE_TIMEOUT, utils.TERMINATE_RETRY_TIMEOUT, + ) + self.assertGreaterEqual(self.mod.TERMINATE_TIMEOUT, 460) names, spared = self._find([ self._cluster('cl-test-mid-run', hours=1), ]) @@ -1815,5 +2492,133 @@ def test_a_since_that_is_not_a_date_is_rejected(self): self.mod.parse_since('last tuesday') +class TestToDatetime(unittest.TestCase): + """ + ``to_datetime`` has to read both timestamp shapes the API returns. + + Most fields come back as RFC 3339, but ``GET /v2/clusters/{id}`` reports + ``expiresAt`` as a Go ``time.Time.String()`` rendering -- verified live: + ``2026-09-17 14:42:41.445984 +0000 UTC`` against a ``createdAt`` of + ``2026-09-17T13:42:41.493848Z`` on the same cluster. The trailing zone name + is not ISO 8601, and parsing it used to fail into ``None``, which reads as + "this cluster never expires". + """ + + def test_rfc_3339(self): + out = to_datetime('2026-09-17T13:42:41.493848Z') + self.assertEqual(out, datetime.datetime(2026, 9, 17, 13, 42, 41, 493848)) + + def test_go_time_string(self): + out = to_datetime('2026-09-17 14:42:41.445984 +0000 UTC') + self.assertEqual(out, datetime.datetime(2026, 9, 17, 14, 42, 41, 445984)) + + def test_offset_is_normalized_to_include_a_colon(self): + # Go writes +0000; datetime.fromisoformat only accepts that spelling on + # 3.11 and later, so the normalizer has to insert the colon itself. This + # asserts on the normalized string rather than on a parsed result + # because the parsed result is only wrong on 3.9 and 3.10, which would + # leave the failure invisible to anyone testing on a newer interpreter. + self.assertEqual( + _normalize_datetime('2026-09-17 14:42:41.445984 +0000 UTC'), + '2026-09-17 14:42:41.445984+00:00', + ) + self.assertEqual( + _normalize_datetime('2026-09-17 09:42:41 +0530 IST'), + '2026-09-17 09:42:41+05:30', + ) + # An offset that already carries a colon is left as it is. + self.assertEqual( + _normalize_datetime('2026-09-17 09:42:41 +05:30 IST'), + '2026-09-17 09:42:41+05:30', + ) + + def test_rfc_3339_fraction_is_padded(self): + # The API trims trailing zeros here too: a job's createdAt came back as + # '2026-09-18T12:39:20.43888Z'. Only 3.11 and later read a fraction that + # is neither 3 nor 6 digits, so before Z was recognized as an offset this + # value skipped the padding and to_datetime_strict raised on 3.10. + self.assertEqual( + _normalize_datetime('2026-09-18T12:39:20.43888Z'), + '2026-09-18T12:39:20.438880+00:00', + ) + self.assertEqual( + to_datetime_strict('2026-09-18T12:39:20.43888Z'), + datetime.datetime(2026, 9, 18, 12, 39, 20, 438880), + ) + + def test_rfc_3339_nanoseconds_are_truncated(self): + # Nine digits does not fit a datetime; the extra ones are dropped. + self.assertEqual( + _normalize_datetime('2026-09-18T12:39:20.438880123Z'), + '2026-09-18T12:39:20.438880+00:00', + ) + + def test_go_time_string_with_truncated_fraction(self): + # Go trims trailing zeros, so the fraction is not always 6 digits. + out = to_datetime('2026-09-17 14:42:41.4 +0000 UTC') + self.assertEqual(out, datetime.datetime(2026, 9, 17, 14, 42, 41, 400000)) + + def test_go_time_string_with_monotonic_reading(self): + out = to_datetime( + '2026-09-17 14:42:41.445984 +0000 UTC m=+0.000000001', + ) + self.assertEqual(out, datetime.datetime(2026, 9, 17, 14, 42, 41, 445984)) + + def test_offset_is_applied_and_dropped(self): + # Shifted onto UTC and left naive, matching the RFC 3339 values, so two + # timestamps read off one object can be compared. + out = to_datetime('2026-09-17 09:42:41 -0500 EST') + self.assertEqual(out, datetime.datetime(2026, 9, 17, 14, 42, 41)) + self.assertIsNone(out.tzinfo) + + def test_both_shapes_subtract(self): + created = to_datetime('2026-09-17T13:42:41.493848Z') + expires = to_datetime('2026-09-17 14:42:41.445984 +0000 UTC') + self.assertAlmostEqual( + (expires - created).total_seconds(), 3600, delta=1, + ) + + def test_date_only(self): + out = to_datetime('2026-09-17') + self.assertEqual(out, datetime.datetime(2026, 9, 17)) + + def test_zero_sentinel_and_unparseable_are_none(self): + self.assertIsNone(to_datetime('0001-01-01T00:00:00Z')) + self.assertIsNone(to_datetime(None)) + self.assertIsNone(to_datetime('')) + self.assertIsNone(to_datetime('not a date')) + + def test_the_go_spelling_of_the_zero_sentinel_is_none_too(self): + # Go's zero time means "unset" -- an expiresAt on a resource that does + # not expire -- and arrives in whichever shape the field uses. Reading + # the Go spelling as a real timestamp reported year 1 as an expiry. + self.assertIsNone(to_datetime('0001-01-01 00:00:00 +0000 UTC')) + # Recognized from the parsed value, so the trimmings Go may add do not + # each need their own literal. + self.assertIsNone( + to_datetime('0001-01-01 00:00:00 +0000 UTC m=+0.000000001'), + ) + self.assertIsNone(to_datetime('0001-01-01 00:00:00 +0000 GMT')) + self.assertIsNone(to_datetime('0001-01-01')) + + def test_datetime_passes_through(self): + given = datetime.datetime(2026, 9, 17, 13, 42, 41) + self.assertIs(to_datetime(given), given) + + def test_strict_reads_the_go_shape_too(self): + out = to_datetime_strict('2026-09-17 14:42:41.445984 +0000 UTC') + self.assertEqual(out, datetime.datetime(2026, 9, 17, 14, 42, 41, 445984)) + + def test_strict_still_raises_on_nothing(self): + with self.assertRaises(TypeError): + to_datetime_strict(None) + with self.assertRaises(ValueError): + to_datetime_strict('0001-01-01T00:00:00Z') + + def test_strict_raises_on_the_go_spelling_of_the_sentinel(self): + with self.assertRaises(ValueError): + to_datetime_strict('0001-01-01 00:00:00 +0000 UTC') + + if __name__ == '__main__': unittest.main() diff --git a/singlestoredb/tests/test_management_v1.py b/singlestoredb/tests/test_management_v1.py index 3328969a..63e7b9dd 100755 --- a/singlestoredb/tests/test_management_v1.py +++ b/singlestoredb/tests/test_management_v1.py @@ -35,6 +35,7 @@ from singlestoredb.management.job import TargetType from singlestoredb.management.region import Region from singlestoredb.management.utils import NamedList +from singlestoredb.tests import utils TEST_DIR = pathlib.Path(os.path.dirname(__file__)) @@ -77,15 +78,25 @@ def setUpClass(cls): region=random.choice(us_regions).id, admin_password=cls.password, firewall_ranges=['0.0.0.0/0'], + expires_at=utils.DEPLOYMENT_EXPIRES_AT, ) try: + # No expiry of its own: only the group has an expiresAt, and it + # takes its workspaces with it. See utils.DEPLOYMENT_EXPIRES_AT. cls.workspace = cls.workspace_group.create_workspace( f'ws-test-{name}-x', wait_on_active=True, ) except Exception: - cls.workspace_group.terminate(force=True) + # Guarded: an unguarded terminate here replaces the create failure + # with whatever the DELETE raised, which both hides the real error + # and leaves the group live with nothing having reported why. + # utils.cleanup_tracked retries it and says so. + try: + cls.workspace_group.terminate(force=True) + except Exception: + pass raise @classmethod @@ -375,6 +386,7 @@ def setUpClass(cls): region=random.choice(us_regions).id, admin_password=cls.password, firewall_ranges=['0.0.0.0/0'], + expires_at=utils.DEPLOYMENT_EXPIRES_AT, ) @classmethod @@ -937,18 +949,30 @@ def test_get_secret(self): except s2.ManagementError: pass - self.manager._post( + created = self.manager._post( 'secrets', json=dict( name='secret_name', value='secret_value', ), - ) - - secret = self.manager.organizations.current.get_secret('secret_name') + ).json() + + # The ID comes from the create response rather than from the lookup + # under test: binding it inside the try would leave the cleanup raising + # UnboundLocalError over whatever the lookup actually failed with. + # Without this the secret outlived every run -- it was only ever + # removed opportunistically by the sweep at the top of the *next* one. + # test_management_v2.py's twin already does it this way. + secret_id = created['secret']['secretID'] + try: + secret = self.manager.organizations.current.get_secret( + 'secret_name', + ) - assert secret.name == 'secret_name' - assert secret.value == 'secret_value' + assert secret.name == 'secret_name' + assert secret.value == 'secret_value' + finally: + self.manager._delete(f'secrets/{secret_id}') @pytest.mark.management @@ -974,15 +998,25 @@ def setUpClass(cls): region=random.choice(us_regions).id, admin_password=cls.password, firewall_ranges=['0.0.0.0/0'], + expires_at=utils.DEPLOYMENT_EXPIRES_AT, ) try: + # No expiry of its own: only the group has an expiresAt, and it + # takes its workspaces with it. See utils.DEPLOYMENT_EXPIRES_AT. cls.workspace = cls.workspace_group.create_workspace( f'ws-test-{name}-x', wait_on_active=True, ) except Exception: - cls.workspace_group.terminate(force=True) + # Guarded: an unguarded terminate here replaces the create failure + # with whatever the DELETE raised, which both hides the real error + # and leaves the group live with nothing having reported why. + # utils.cleanup_tracked retries it and says so. + try: + cls.workspace_group.terminate(force=True) + except Exception: + pass raise @classmethod diff --git a/singlestoredb/tests/test_management_v2.py b/singlestoredb/tests/test_management_v2.py index 5842bdc5..fe65e47d 100644 --- a/singlestoredb/tests/test_management_v2.py +++ b/singlestoredb/tests/test_management_v2.py @@ -1332,6 +1332,7 @@ def setUpClass(cls): region=region, size='S-00', firewall_ranges=['0.0.0.0/0'], + expires_at=utils.DEPLOYMENT_EXPIRES_AT, project=_project_id(cls.manager), wait_on_active=True, ) diff --git a/singlestoredb/tests/utils.py b/singlestoredb/tests/utils.py index 94d754c4..09f8ba3c 100644 --- a/singlestoredb/tests/utils.py +++ b/singlestoredb/tests/utils.py @@ -2,6 +2,7 @@ # type: ignore """Utilities for testing.""" import glob +import json import logging import os import random @@ -306,6 +307,38 @@ def drop_user(name: str) -> None: # ignored -- so tracked objects do not have to be untracked by the tests that # clean up after themselves. # +# Everything above is in-process, which is the one thing it cannot fix: the +# sweep, the ledger and `tearDownClass` all die with the interpreter. A job +# killed mid-provision -- GitHub force-terminates a cancelled job's remaining +# steps after a five-minute cancellation timeout -- leaves a PENDING cluster +# that no code here will ever get another chance to delete. `expires_at` is +# the answer to that, and only that: it is a property of the deployment, so +# the control plane honours it whether or not this process is still alive. +# + +#: Expiry to request on every deployment a test creates, as the duration +#: string `POST` accepts (`resources/create_test_cluster.py` has passed one +#: nightly since it was written). The backstop under the sweep and the ledger, +#: not a replacement for either: a test still terminates what it created, and +#: nothing waits for an expiry to fire. +#: +#: Two hours, against a `wait_timeout` of 1200s and a longest test (Fusion +#: `CREATE`/`DROP`, which provisions twice in sequence) of about twenty +#: minutes. Enough headroom that an expiry can never land on a deployment a +#: test is still using, which would show up as an unrelated flake and be read +#: as an API fault. +#: +#: Applied to what accepts it, which is the deployments that cost: v2 +#: `ClusterManager.create_cluster` and v1 +#: `WorkspaceManager.create_workspace_group`. The rest take no `expires_at` and +#: need none: +#: +#: * a v1 workspace -- `expiresAt` is a property of the group, and terminating +#: the group takes its workspaces with it; +#: * the starter deployments -- `create_starter_cluster` and +#: `create_starter_workspace` have no such argument, and being shared tier +#: they are not what a leak costs. +DEPLOYMENT_EXPIRES_AT = '2h' #: (owner, label, object) for every deployment created so far and not yet #: swept. The owner is the test class that was running at creation time, so @@ -336,6 +369,160 @@ def set_owner(owner: str) -> None: _owner = owner +# +# Durable deployment ledger +# +# Everything above this point is in-memory only, and that is the one leak the +# sweeps cannot cover. A cancelled CI job is the proven case: GH Actions run +# 35631802648, job ``test-coverage``, was cancelled 19 minutes into +# ``TestClusterFusion.setUpClass``'s ``create_cluster(wait_on_active=True, +# wait_timeout=1200)``. The log ends at ``##[error]The operation was +# canceled.`` with no pytest summary, no "Terminated deployments left behind +# by tests:" and no ``STILL LIVE`` banner -- the process never got to sweep, +# and GitHub's cancellation grace period is nowhere near long enough for +# pytest to unwind three nested class fixtures, list to recover three +# in-flight creations and issue three DELETEs. After the SIGKILL that follows, +# ``_tracked`` and ``_in_flight`` are gone with the process and *nothing on +# disk* records that three clusters were created. +# +# So every creation is also appended to a JSONL file, flushed and fsync'd per +# line, which ``cleanup_deployments.py --ledger`` reads afterwards from a +# separate process -- an ``if: always()`` CI step that still runs on +# cancellation. The ledger is the record; the in-memory sweeps stay exactly as +# they were and remain the fast path. +# +# Opt-in, via SINGLESTOREDB_TEST_DEPLOYMENT_LOG. With the variable unset +# nothing is written and behaviour is byte-for-byte what it was: a local run +# has a human watching it and does not need a file to reap from. +# + +#: Ledger event kinds, in the order a deployment normally produces them: +#: +#: * ``pending`` -- the creator is about to be called. Written *before* the +#: POST, from the name argument, because the whole point is the window where +#: the server has a billable deployment and this process has no id for it. +#: * ``live`` -- the creation returned (or an orphan was recovered), so there +#: is an id. +#: * ``gone`` -- it has been terminated. +#: +#: The reaper folds the file: anything whose last event is not ``gone`` is +#: still live. A ``pending`` with no matching ``live`` is the cancelled-mid- +#: wait case, and it is resolved by name rather than by id. + +#: Environment variable naming the ledger file. Read per write rather than +#: cached at import so a test can point it at a tmp_path with +#: ``mock.patch.dict(os.environ, ...)``. +LEDGER_ENV_VAR = 'SINGLESTOREDB_TEST_DEPLOYMENT_LOG' + +#: Deployment kind for each created object's class. The ledger records a kind +#: so the reaper knows which manager and which point lookup to resolve a +#: record against, instead of guessing from the name -- ``cl-test-abc`` and +#: ``ws-test-abc`` are only distinguishable by convention, and a ``--ledger`` +#: run deliberately does not consult :data:`cleanup_deployments.PATTERNS`. +#: +#: Keyed by class name rather than by the class itself to avoid importing v1 +#: and v2 management just to write a log line. +_KIND_BY_CLASS = { + 'WorkspaceGroup': 'workspace_group', + 'Workspace': 'workspace', + 'StarterWorkspace': 'starter_workspace', + 'Cluster': 'cluster', + 'StarterCluster': 'starter_cluster', +} + + +def ledger_path() -> Optional[str]: + """Path of the deployment ledger, or None if none was configured.""" + return os.environ.get(LEDGER_ENV_VAR) or None + + +def _ledger_write(**record: Any) -> None: + """ + Append one record to the deployment ledger. + + Opened, written and closed per record, with ``flush()`` and ``os.fsync()`` + before the handle goes: surviving SIGKILL is the entire purpose, and a + line still sitting in a buffer when the process dies records nothing. The + cost is one open per creation, against a creation that takes minutes. + + ``O_APPEND`` plus one ``write()`` per line is what makes this safe for the + parallel default (``-n 2``): the xdist workers are separate processes + sharing the file, and a single write of well under PIPE_BUF cannot + interleave with another's on Linux. No locking, therefore, and no partial + lines for the reaper to choke on. + + Never raises. This sits on the creation path of every management test, so + a full disk or an unwritable path must cost a warning, not a test failure. + """ + path = ledger_path() + if not path: + return + try: + # default=str so an unexpected value (a datetime, an enum) degrades to + # its repr instead of raising and losing the whole record. + line = json.dumps(record, default=str, sort_keys=True) + '\n' + with open(path, 'a', encoding='utf-8') as file: + file.write(line) + file.flush() + os.fsync(file.fileno()) + except Exception as exc: + logger.warning( + f'Could not append {record!r} to the deployment ledger at ' + f'{path!r}; a deployment this run creates may not be reaped: ' + f'{exc}', + ) + + +def _ledger_kind(obj: Any) -> Optional[str]: + """Ledger kind for a created object, or None if it is not a deployment.""" + return _KIND_BY_CLASS.get(type(obj).__name__) + + +def ledger_pending(kind: str, args: Tuple[Any, ...], kwargs: Any) -> None: + """ + Record that a deployment of this kind is about to be created. + + The name is taken the same way :func:`_recover_orphan` takes it -- keyword + first, else the first positional -- because it is the first parameter of + every creator, which ``test_management_utils.py`` pins. A record with no + usable name is skipped: there would be nothing for the reaper to resolve. + """ + name = kwargs.get('name') or (args[0] if args else None) + if not isinstance(name, str): + return + _ledger_write(event='pending', kind=kind, name=name) + + +def ledger_live(obj: Any) -> None: + """Record that a created deployment exists, now that it has an id.""" + kind = _ledger_kind(obj) + if kind is None: + return + _ledger_write( + event='live', kind=kind, + id=getattr(obj, 'id', None), + name=getattr(obj, 'name', None), + ) + + +def ledger_gone(obj: Any) -> None: + """ + Record that a deployment has been terminated. + + Carries the name as well as the id so it also cancels a ``pending`` + record: an orphan recovered by name and then swept in-process would + otherwise still be listed as live by the reaper. + """ + kind = _ledger_kind(obj) + if kind is None: + return + _ledger_write( + event='gone', kind=kind, + id=getattr(obj, 'id', None), + name=getattr(obj, 'name', None), + ) + + def _is_mocked(obj: Any) -> bool: """ Did this object come out of a mocked manager? @@ -378,6 +565,10 @@ def track(obj: Any, label: str = '') -> Any: ), obj, )) + # Here rather than in the wrapper, so an orphan that `_recover_orphan` + # digs out of a listing gets an id into the ledger too -- that path + # reaches the server only through this function. + ledger_live(obj) return obj @@ -427,24 +618,111 @@ def _recover_orphan( def untrack(obj: Any) -> None: """Forget a deployment that has been terminated.""" + found = False for i, entry in reversed(list(enumerate(_tracked))): if entry[2] is obj: _tracked.pop(i) - - -def terminate(obj: Any) -> None: + found = True + # Only for something that was actually tracked: untracking an object that + # was never registered -- a mocked one, or one already swept -- says + # nothing about whether a real deployment is gone, and a spurious ``gone`` + # would hide a live cluster from the reaper. + if found: + ledger_gone(obj) + + +#: How long :func:`terminate` keeps retrying a deployment the API will not +#: delete yet, and how long it waits between attempts. Three minutes at 15s +#: spacing: the case being covered is a deployment killed mid-provision, which +#: has to finish coming up before it can be torn down, and an S-00 cluster +#: reaching ACTIVE is ~460s at worst. Waiting the full provision out here would +#: stall the sweep between every test class, so this buys the common case -- +#: a deployment most of the way up -- and leaves the rest to the end-of-session +#: sweep and then to ``cleanup_deployments.py``, which is the end of the line +#: and waits out a full provision with a longer budget of its own +#: (``cleanup_deployments.TERMINATE_TIMEOUT``). +TERMINATE_RETRY_TIMEOUT = 180.0 +TERMINATE_RETRY_INTERVAL = 15.0 + + +def _terminate_once(obj: Any) -> None: """ - Terminate a deployment, whatever kind it is. + Issue one terminate, whatever this kind's signature looks like. ``force=True`` is what makes a workspace group with live workspaces in it - go away; the starter variants take no arguments at all. + go away; the starter variants (``StarterWorkspace.terminate``, + ``StarterCluster.terminate``) take no arguments at all. + + The signature is inspected rather than discovered by catching ``TypeError`` + from the call, as this used to do. That ``except TypeError`` also caught a + ``TypeError`` raised from *inside* a terminate that did accept ``force``, + and then retried without it -- two DELETEs for one deployment, the second + of them not forced, which is the one shape that leaves a workspace group + behind. """ + import inspect + try: + params = inspect.signature(obj.terminate).parameters + except (TypeError, ValueError): # pragma: no cover - unintrospectable + # A builtin or a C-level callable. Fall back to the old behaviour. + params = {} + + if 'force' in params: obj.terminate(force=True) - except TypeError: + else: obj.terminate() +def terminate( + obj: Any, + timeout: float = TERMINATE_RETRY_TIMEOUT, + interval: float = TERMINATE_RETRY_INTERVAL, +) -> None: + """ + Terminate a deployment, whatever kind it is, retrying a 4xx refusal. + + A deployment killed mid-provision is ``PENDING``/``TRANSITIONING``, and the + API refuses to delete it in that state with a 400 or a 409. Nothing retries + that: ``Manager.RETRY_STATUSES`` is ``{429, 500, 502, 503, 504}`` + (``management/manager.py:49``), urllib3 only retries what is in that list, + and there is no wait-until-deletable helper anywhere in the SDK. So the + per-class sweep logged a warning, the session-end sweep tried exactly once + more -- usually still too early -- and the deployment was left running. + + Hence the bounded retry here. Only 4xx other than 404 is retried: + + * 404 means it is already gone, so retrying would burn the whole budget + waiting for something that will never come back. Re-raised, as before, + which is also what ``_is_gone()`` upstream normally prevents. + * 5xx and 429 are already retried inside the transport, so seeing one here + means the transport gave up; another round trip from this layer is not + what fixes it. + + Raises the last error if the budget runs out, so ``cleanup_tracked()`` + keeps the deployment tracked and the session-end sweep gets another go. + """ + import time + + deadline = time.monotonic() + timeout + while True: + try: + _terminate_once(obj) + return + except ManagementError as exc: + errno = exc.errno + if errno is None or errno == 404 or not 400 <= errno < 500: + raise + # No budget left for another attempt *plus* the wait before it. + if time.monotonic() + interval >= deadline: + raise + logger.info( + f'{obj!r} is not deletable yet ({exc}); retrying the ' + f'terminate in {interval:g}s', + ) + time.sleep(interval) + + def _creator_is_mocked(target: Any) -> bool: """ Is this creation call going through a mocked manager? @@ -473,9 +751,15 @@ def _creator_is_mocked(target: Any) -> bool: ) -#: (module, class, method, finder) tuples for the calls that bring a billable -#: deployment into existence. Wrapping them is what makes tracking automatic, -#: so a new test cannot leak a cluster by forgetting to register it. +#: (module, class, method, kind, finder) tuples for the calls that bring a +#: billable deployment into existence. Wrapping them is what makes tracking +#: automatic, so a new test cannot leak a cluster by forgetting to register it. +#: +#: ``kind`` is the ledger kind the call produces, and must be a value of +#: :data:`_KIND_BY_CLASS`: it is what lets the ``pending`` record -- written +#: before the POST, when nothing has an id yet -- say which manager the reaper +#: should search. It is stated here rather than derived from ``method_name`` +#: because ``create_workspace`` appears twice, on two different receivers. #: #: ``finder`` takes the receiver -- the manager, or the group for #: ``WorkspaceGroup.create_workspace`` -- and returns the collection to search @@ -484,34 +768,34 @@ def _creator_is_mocked(target: Any) -> bool: _CREATORS = [ ( 'singlestoredb.management.v1.workspace', 'WorkspaceManager', - 'create_workspace_group', + 'create_workspace_group', 'workspace_group', lambda recv: recv.workspace_groups, ), ( 'singlestoredb.management.v1.workspace', 'WorkspaceManager', - 'create_workspace', + 'create_workspace', 'workspace', # WorkspaceManager has no `workspaces` of its own, so the search goes # group by group. Only ever walked on the failure path. lambda recv: [w for g in recv.workspace_groups for w in g.workspaces], ), ( 'singlestoredb.management.v1.workspace', 'WorkspaceManager', - 'create_starter_workspace', + 'create_starter_workspace', 'starter_workspace', lambda recv: recv.starter_workspaces, ), ( 'singlestoredb.management.v1.workspace', 'WorkspaceGroup', - 'create_workspace', + 'create_workspace', 'workspace', lambda recv: recv.workspaces, ), ( 'singlestoredb.management.v2.cluster', 'ClusterManager', - 'create_cluster', + 'create_cluster', 'cluster', lambda recv: recv.clusters, ), ( 'singlestoredb.management.v2.cluster', 'ClusterManager', - 'create_starter_cluster', + 'create_starter_cluster', 'starter_cluster', lambda recv: recv.starter_clusters, ), ] @@ -519,7 +803,7 @@ def _creator_is_mocked(target: Any) -> bool: _tracking_installed = False -def _tracking_wrapper(func: Any, finder: Any) -> Any: +def _tracking_wrapper(func: Any, kind: str, finder: Any) -> Any: """ Wrap a creation method so its result -- or its orphan -- gets tracked. @@ -546,6 +830,14 @@ def _tracking_wrapper(func: Any, finder: Any) -> Any: unit test's stubbed ``get_cluster`` returns. Nothing a mocked creator returns names a deployment that exists, so the receiver's verdict is the authoritative one and it is the one used here. + + The ``pending`` ledger record is written here, and deliberately *before* + ``func`` is called rather than after: from the moment the creator POSTs + there is a billable deployment, and everything that could record it -- + ``track()`` on return, ``_recover_orphan()`` in the ``except``, + ``recover_in_flight()`` from a signal handler -- runs after the wait that + a cancelled CI job never survives. A ``pending`` line on disk is the only + thing that outlives a SIGKILL there. """ import functools @@ -555,6 +847,7 @@ def wrapper(receiver: Any, *args: Any, **kwargs: Any) -> Any: entry = (receiver, finder, args, kwargs) if not mocked: _in_flight.append(entry) + ledger_pending(kind, args, kwargs) try: out = func(receiver, *args, **kwargs) return out if mocked else track(out) @@ -628,12 +921,12 @@ def install_deployment_tracking() -> None: import importlib - for module_name, class_name, method_name, finder in _CREATORS: + for module_name, class_name, method_name, kind, finder in _CREATORS: try: klass = getattr(importlib.import_module(module_name), class_name) setattr( klass, method_name, - _tracking_wrapper(getattr(klass, method_name), finder), + _tracking_wrapper(getattr(klass, method_name), kind, finder), ) except AttributeError as exc: # A renamed method must not silently stop being tracked. @@ -708,6 +1001,11 @@ def cleanup_tracked(owner: Optional[str] = None) -> List[str]: _, label, obj = entry if _is_gone(obj): _tracked.remove(entry) + # The server says it is gone, which is exactly what the ledger's + # ``gone`` means -- a test that terminated in its own teardown gets + # its record closed here rather than leaving the reaper to look up + # an id that 404s. + ledger_gone(obj) continue try: terminate(obj) @@ -719,6 +1017,7 @@ def cleanup_tracked(owner: Optional[str] = None) -> List[str]: logger.warning(f'Could not terminate {label}: {exc}') else: _tracked.remove(entry) + ledger_gone(obj) removed.append(label) return removed @@ -751,10 +1050,16 @@ def tracked_labels() -> List[str]: # does ``TestWorkspaceFusion``, whose workspace groups are the subject of its # ``SHOW WORKSPACE GROUPS`` assertions and cost 40s to deploy unwaited anyway. # -# What makes the four borrowers safe is that each scopes its assertions to -# itself: every Stage path is namespaced with the class's ``cls.id``, job -# listings filter by job id rather than listing a deployment's jobs, and none -# of them asserts a row count over an org-wide listing. +# What makes the borrowers safe is that each scopes its assertions to itself: +# every Stage path is namespaced with the class's ``cls.id``, job listings +# filter by job id rather than listing a deployment's jobs, and none of them +# asserts a row count over an org-wide listing. +# +# ``TestClusterFusion`` is the one borrower that does count rows, because +# ``SHOW CLUSTERS ... LIKE`` is what it tests. It stays inside that rule by +# counting over :func:`shared_cluster_pattern` -- which matches this process's +# pool and nothing else -- against :func:`shared_cluster_names` rather than a +# literal, so growing the pool cannot break it. # # The pool is process-wide, so under ``pytest-xdist`` every worker that gets a # borrowing class builds a pool of its own. The ``xdist_group`` marks below @@ -763,18 +1068,24 @@ def tracked_labels() -> List[str]: #: ``xdist_group`` names for the classes that borrow from the pool, so #: ``--dist loadgroup`` puts each set on one worker and each set builds one -#: pool. Two groups rather than one: a single group serialises all four classes +#: pool. Two groups rather than one: a single group serialises every borrower #: behind one pool build, and the groups run concurrently on separate workers, #: so splitting costs one extra cluster and halves that chain. #: -#: Stage wants two clusters (``TestStageFusion`` names a second one in -#: ``IN GROUP``) and jobs want one, so the split follows what they borrow: +#: The split follows what each set borrows -- three for Stage, one for Jobs: #: -#: * ``SHARED_CLUSTER_STAGE_GROUP`` -- ``TestStageFusion``, v2 ``TestStage`` +#: * ``SHARED_CLUSTER_STAGE_GROUP`` -- ``TestStageFusion`` (two; it names a +#: second in ``IN GROUP``), v2 ``TestStage`` (one), ``TestClusterFusion`` +#: (three, for its ``LIKE``/``ORDER BY``/``LIMIT`` rows) #: * ``SHARED_CLUSTER_JOBS_GROUP`` -- ``TestJobsFusion``, v2 ``TestJob`` #: +#: ``TestClusterFusion`` sits with Stage rather than Jobs deliberately: the +#: pool grows to the largest request, so putting the class that wants three +#: with the group that already wants two costs one extra cluster, where +#: putting it with Jobs would cost two and leave Stage's pool untouched. +#: #: Without ``-n``/``--dist loadgroup`` the marks do nothing: one process, one -#: pool of two, which is the serial behaviour they were added on top of. +#: pool of three, which is the serial behaviour they were added on top of. SHARED_CLUSTER_STAGE_GROUP = 'shared-cluster-stage' SHARED_CLUSTER_JOBS_GROUP = 'shared-cluster-jobs' @@ -860,6 +1171,7 @@ def setUpClass(cls): # pool cluster stands in for those, so it has to be at # least as reachable as what it replaces. firewall_ranges=['0.0.0.0/0'], + expires_at=DEPLOYMENT_EXPIRES_AT, project=project_id, wait_on_active=True, wait_timeout=1200, @@ -871,6 +1183,32 @@ def setUpClass(cls): return _pool[:count] +def shared_cluster_pattern() -> str: + """ + ``LIKE`` pattern matching this process's pool clusters and nothing else. + + The suffix is what scopes it: ``_pool_id`` is minted per process, so a + concurrent run's pool -- or another xdist worker's -- does not match, and + neither does any other ``cl-test-*`` deployment. + + For a suite asserting an exact row count over the pool, pair it with + :func:`shared_cluster_names` rather than a literal: the pool grows to the + largest request any class makes, so the number is not fixed at import. + """ + return f'cl-test-shared-%-{_pool_id}' + + +def shared_cluster_names() -> List[str]: + """ + Names of every cluster in the pool as it stands right now. + + Read at assertion time, not cached: a class that runs later and asks for + more clusters than this one did grows the pool, and an expectation built + from a literal count would go stale the moment that happened. + """ + return [x.name for x in _pool] + + class CountingManager: """ Stand-in for a :class:`Manager` that records every request.